Handle empty stdin in YDB CLI running in a job (#18395)

pnv1 · web-flow · commit d0e7a118c0b8 · 2025-05-21T10:32:47.000+03:00
diff --git a/ydb/apps/ydb/CHANGELOG.md b/ydb/apps/ydb/CHANGELOG.md
@@ -1,3 +1,4 @@
+* Fixed a bug where ydb cli was trying to read parameters from stdin even if it had no data.
 * Add `--replace` option to `ydb tools restore` command. If enabled, scheme objects present in the backup would be dropped before restoring.
 * Added date range parameters (--date-to, --date-from to support uniform PK distribution) for ydb workload log run operations including bulk_upsert, insert, and upsert
 * Do not save to local backups destination tables of `ASYNC REPLICATION` and its changefeeds. It prevents duplication of changefeeds and reduces the amount of space the backup takes on disk.
diff --git a/ydb/public/lib/ydb_cli/commands/ydb_service_scripting.cpp b/ydb/public/lib/ydb_cli/commands/ydb_service_scripting.cpp
@@ -116,7 +116,7 @@ int TCommandExecuteYqlScript::Run(TConfig& config) {
 
         if (!Parameters.empty() || InputParamStream) {
             THolder<TParamsBuilder> paramBuilder;
-            while (GetNextParams(driver, Script, paramBuilder)) {
+            while (GetNextParams(driver, Script, paramBuilder, config.IsVerbose())) {
                 auto asyncResult = client.ExecuteYqlScript(
                         Script,
                         paramBuilder->Build(),
diff --git a/ydb/public/lib/ydb_cli/commands/ydb_service_table.cpp b/ydb/public/lib/ydb_cli/commands/ydb_service_table.cpp
@@ -457,7 +457,7 @@ int TCommandExecuteQuery::ExecuteDataQuery(TConfig& config) {
 
     if (!Parameters.empty() || InputParamStream) {
         THolder<TParamsBuilder> paramBuilder;
-        while (GetNextParams(driver, Query, paramBuilder)) {
+        while (GetNextParams(driver, Query, paramBuilder, config.IsVerbose())) {
             TParams params = paramBuilder->Build();
             auto operation = [this, &txSettings, &params, &settings, &asyncResult](NTable::TSession session) {
                 auto promise = NThreading::NewPromise<NTable::TDataQueryResult>();
@@ -723,7 +723,7 @@ int TCommandExecuteQuery::ExecuteQueryImpl(TConfig& config) {
     SetInterruptHandlers();
     if (!Parameters.empty() || InputParamStream) {
         THolder<TParamsBuilder> paramBuilder;
-        while (GetNextParams(driver, Query, paramBuilder)) {
+        while (GetNextParams(driver, Query, paramBuilder, config.IsVerbose())) {
             auto operation = [this, &paramBuilder, &settings, &asyncResult](TClient client) {
                 auto promise = NThreading::NewPromise<TPartIterator<TClient>>();
                 asyncResult = promise.GetFuture();
diff --git a/ydb/public/lib/ydb_cli/commands/ydb_sql.cpp b/ydb/public/lib/ydb_cli/commands/ydb_sql.cpp
@@ -180,7 +180,7 @@ int TCommandSql::RunCommand(TConfig& config) {
     if (!Parameters.empty() || InputParamStream) {
         // Execute query with parameters
         THolder<TParamsBuilder> paramBuilder;
-        while (!IsInterrupted() && GetNextParams(driver, Query, paramBuilder)) {
+        while (!IsInterrupted() && GetNextParams(driver, Query, paramBuilder, config.IsVerbose())) {
             auto asyncResult = client.StreamExecuteQuery(
                     Query,
                     NQuery::TTxControl::NoTx(),
diff --git a/ydb/public/lib/ydb_cli/commands/ydb_yql.cpp b/ydb/public/lib/ydb_cli/commands/ydb_yql.cpp
@@ -101,7 +101,7 @@ int TCommandYql::RunCommand(TConfig& config, const TString& script) {
 
     if (!Parameters.empty() || InputParamStream) {
         THolder<TParamsBuilder> paramBuilder;
-        while (!IsInterrupted() && GetNextParams(driver, Script, paramBuilder)) {
+        while (!IsInterrupted() && GetNextParams(driver, Script, paramBuilder, config.IsVerbose())) {
             auto asyncResult = client.StreamExecuteYqlScript(
                     script,
                     paramBuilder->Build(),
diff --git a/ydb/public/lib/ydb_cli/common/parameters.cpp b/ydb/public/lib/ydb_cli/common/parameters.cpp
@@ -192,6 +192,53 @@ void TCommandWithParameters::AddParams(TParamsBuilder& paramBuilder) {
     }
 }
 
+namespace {
+    bool StdinHasData(bool verbose) {
+#if defined(_win32_)
+        // Too complex case for Windows
+        return false;
+#else
+        // fd_set to store a set of descriptor set.
+        fd_set read_fds;
+        FD_ZERO(&read_fds);
+        FD_SET(STDIN_FILENO, &read_fds);
+
+        struct timeval timeout;
+        timeout.tv_sec = 0;
+        timeout.tv_usec = 0; // No timeout, instant check
+
+        // Check if stdin is available for reading
+        int selectResult = select(STDIN_FILENO + 1, &read_fds, NULL, NULL, &timeout);
+        if (selectResult == 0) {
+            if (verbose) {
+                Cerr << "stdin is not available" << Endl;
+            }
+            return false;
+        }
+
+        // Trying to read 1 symbol from stdin
+        char buffer[1];
+        ssize_t result = read(fileno(stdin), buffer, sizeof(buffer));
+        if (result == -1) {
+            if (verbose) {
+                Cerr << "Error reading from stdin. Error: " << strerror(errno) << Endl;
+            }
+        } else if (result == 0) {
+            if (verbose) {
+                Cerr << "No data from stdin" << Endl;
+            }
+        } else {
+            if (verbose) {
+                Cerr << "stdin has data, returning first symbol '" << buffer[0] << "' back..." << Endl;
+            }
+            ungetc(buffer[0], stdin);
+            return true;
+        }
+        return false;
+#endif
+    }
+}
+
 void TCommandWithParameters::ParseParameters(TClientCommand::TConfig& config) {
     // Deprecated options with defaults:
     if (!DeprecatedSkipRows.empty()) {
@@ -253,10 +300,12 @@ void TCommandWithParameters::ParseParameters(TClientCommand::TConfig& config) {
         }
     }
 
+    bool verbose = config.IsVerbose();
+
     if (InputFiles.empty()) {
-        if (!IsStdinInteractive() && !ReadingSomethingFromStdin) {
+        if (!IsStdinInteractive() && !ReadingSomethingFromStdin && StdinHasData(verbose)) {
             // By default reading params from stdin
-            SetParamsInputFromStdin();
+            SetParamsInputFromStdin(verbose);
         }
     } else {
         auto& file = InputFiles[0];
@@ -265,10 +314,10 @@ void TCommandWithParameters::ParseParameters(TClientCommand::TConfig& config) {
                 throw TMisuseException() << "Path to input file is \"-\", meaning that parameter value[s] should be read "
                     "from stdin. This is only available in non-interactive mode";
             }
-            SetParamsInputFromStdin();
+            SetParamsInputFromStdin(verbose);
         } else {
             if (IsStdinInteractive() || ReadingSomethingFromStdin) {
-                SetParamsInputFromFile(file);
+                SetParamsInputFromFile(file, verbose);
             } else {
                 throw TMisuseException() << "Path to input file is \"" << file << "\", meaning that parameter value[s]"
                     " should be read from file. This is only available in interactive mode. Can't read parameters both"
@@ -320,21 +369,27 @@ void TCommandWithParameters::SetParamsInput(IInputStream* input) {
     }
 }
 
-void TCommandWithParameters::SetParamsInputFromStdin() {
+void TCommandWithParameters::SetParamsInputFromStdin(bool verbose) {
     if (ReadingSomethingFromStdin) {
         throw TMisuseException() << "Can't read both parameters and query text from stdinput";
     }
     ReadingSomethingFromStdin = true;
     SetParamsInput(&Cin);
+    if (verbose) {
+        Cerr << "Reading parameters from stdin" << Endl;
+    }
 }
 
-void TCommandWithParameters::SetParamsInputFromFile(TString& file) {
+void TCommandWithParameters::SetParamsInputFromFile(TString& file, bool verbose) {
     TFsPath fsPath = GetExistingFsPath(file, "input file");
     InputFileHolder = MakeHolder<TFileInput>(fsPath);
     SetParamsInput(InputFileHolder.Get());
+    if (verbose) {
+        Cerr << "Reading parameters from file \"" << file << '\"' << Endl;
+    }
 }
 
-void TCommandWithParameters::InitParamTypes(const TDriver& driver, const TString& queryText) {
+void TCommandWithParameters::InitParamTypes(const TDriver& driver, const TString& queryText, bool verbose) {
     if (SyntaxType == NQuery::ESyntax::Pg) {
         ParamTypes.clear();
         return;
@@ -343,9 +398,15 @@ void TCommandWithParameters::InitParamTypes(const TDriver& driver, const TString
     auto types = TYqlParamParser::GetParamTypes(queryText);
     if (types.has_value()) {
         ParamTypes = *types;
+        if (verbose) {
+            Cerr << "Successfully retrieved parameter types from query text locally" << Endl;
+        }
         return;
     }
 
+    if (verbose) {
+        Cerr << "Failed to retrieve parameter types from query text locally. Executing ExplainYqlScript..." << Endl;
+    }
     // Fallback to ExplainYql
     NScripting::TScriptingClient client(driver);
     auto explainSettings = NScripting::TExplainYqlRequestSettings()
@@ -360,11 +421,12 @@ void TCommandWithParameters::InitParamTypes(const TDriver& driver, const TString
     ParamTypes = result.GetParameterTypes();
 }
 
-bool TCommandWithParameters::GetNextParams(const TDriver& driver, const TString& queryText, THolder<TParamsBuilder>& paramBuilder) {
+bool TCommandWithParameters::GetNextParams(const TDriver& driver, const TString& queryText,
+        THolder<TParamsBuilder>& paramBuilder, bool verbose) {
     paramBuilder = MakeHolder<TParamsBuilder>();
     if (IsFirstEncounter) {
         IsFirstEncounter = false;
-        InitParamTypes(driver, queryText);
+        InitParamTypes(driver, queryText, verbose);
 
         if (!InputParamStream) {
             AddParams(*paramBuilder);
diff --git a/ydb/public/lib/ydb_cli/common/parameters.h b/ydb/public/lib/ydb_cli/common/parameters.h
@@ -32,7 +32,7 @@ class TCommandWithParameters : public TCommandWithExamples, public TCommandWithI
     void AddLegacyBatchParametersOptions(TClientCommand::TConfig& config);
     void AddDefaultParamFormats(TClientCommand::TConfig& config);
     void AddLegacyStdinFormats(TClientCommand::TConfig& config);
-    bool GetNextParams(const TDriver& driver, const TString& queryText, THolder<TParamsBuilder>& paramBuilder);
+    bool GetNextParams(const TDriver& driver, const TString& queryText, THolder<TParamsBuilder>& paramBuilder, bool verbose);
     
     THashMap<EDataFormat, TString>& GetInputFormatDescriptions() override;
 
@@ -41,9 +41,9 @@ class TCommandWithParameters : public TCommandWithExamples, public TCommandWithI
     static void ParseJson(TString&& str, std::map<TString, TString>& result);
     void ApplyJsonParams(const std::map<TString, TString>& params, TParamsBuilder& paramBuilder);
     void SetParamsInput(IInputStream* input);
-    void SetParamsInputFromFile(TString& file);
-    void SetParamsInputFromStdin();
-    void InitParamTypes(const TDriver& driver, const TString& queryText);
+    void SetParamsInputFromFile(TString& file, bool verbose);
+    void SetParamsInputFromStdin(bool verbose);
+    void InitParamTypes(const TDriver& driver, const TString& queryText, bool verbose);
 
     TMaybe<TString> ReadData();
 

Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,4 @@`
	`1`	`+* Fixed a bug where ydb cli was trying to read parameters from stdin even if it had no data.`
`1`	`2`	* Add `--replace` option to `ydb tools restore` command. If enabled, scheme objects present in the backup would be dropped before restoring.
`2`	`3`	`* Added date range parameters (--date-to, --date-from to support uniform PK distribution) for ydb workload log run operations including bulk_upsert, insert, and upsert`
`3`	`4`	* Do not save to local backups destination tables of `ASYNC REPLICATION` and its changefeeds. It prevents duplication of changefeeds and reduces the amount of space the backup takes on disk.