Improve csv import, steps 1 & 2 (#11679)

Author: pnv1

ydb/public/lib/ydb_cli/commands/ydb_service_import.cpp

@@ -302,6 +302,7 @@ int TCommandImportFromCsv::Run(TConfig& config) {
     settings.NewlineDelimited(NewlineDelimited);
     settings.HeaderRow(HeaderRow);
     settings.NullValue(NullValue);
+    settings.Verbose(config.IsVerbose());
 
     if (Delimiter.size() != 1) {
         throw TMisuseException()

ydb/public/lib/ydb_cli/common/csv_parser.cpp

@@ -1,5 +1,6 @@
 #include "csv_parser.h"
 
+#include <ydb/public/api/protos/ydb_value.pb.h>
 #include <ydb/public/lib/ydb_cli/common/common.h>
 
 #include <library/cpp/string_utils/csv/csv.h>
@@ -177,7 +178,7 @@ class TCsvToYdbConverter {
         }
     }
 
-    void BuildValue(TStringBuf token) {
+    void BuildValue(const TStringBuf& token) {
         switch (Parser.GetKind()) {
         case TTypeParser::ETypeKind::Primitive: {
             BuildPrimitive(TString(token));
@@ -279,7 +280,7 @@ class TCsvToYdbConverter {
         throw TCsvParseException() << "Expected bool value: \"true\" or \"false\", received: \"" << token << "\".";
     }
 
-    void EnsureNull(TStringBuf token) const {
+    void EnsureNull(const TStringBuf& token) const {
         if (!NullValue) {
             throw TCsvParseException() << "Expected null value instead of \"" << token << "\", but null value is not set.";
         }
@@ -288,7 +289,7 @@ class TCsvToYdbConverter {
         }
     }
 
-    TValue Convert(TStringBuf token) {
+    TValue Convert(const TStringBuf& token) {
         BuildValue(token);
         return Builder.Build();
     }
@@ -317,10 +318,10 @@ TCsvParseException FormatError(const std::exception& inputError,
 }
 
 TValue FieldToValue(TTypeParser& parser,
-                    TStringBuf token,
+                    const TStringBuf& token,
                     const std::optional<TString>& nullValue,
                     const TCsvParser::TParseMetadata& meta,
-                    TString columnName) {
+                    const TString& columnName) {
     try {
         TCsvToYdbConverter converter(parser, nullValue);
         return converter.Convert(token);
@@ -331,7 +332,7 @@ TValue FieldToValue(TTypeParser& parser,
 
 TStringBuf Consume(NCsvFormat::CsvSplitter& splitter,
                    const TCsvParser::TParseMetadata& meta,
-                   TString columnName) {
+                   const TString& columnName) {
     try {
         return splitter.Consume();
     } catch (std::exception& e) {
@@ -342,30 +343,30 @@ TStringBuf Consume(NCsvFormat::CsvSplitter& splitter,
 }
 
 TCsvParser::TCsvParser(TString&& headerRow, const char delimeter, const std::optional<TString>& nullValue,
-                       const std::map<TString, TType>* paramTypes,
+                       const std::map<TString, TType>* destinationTypes,
                        const std::map<TString, TString>* paramSources) 
     : HeaderRow(std::move(headerRow))
     , Delimeter(delimeter)
     , NullValue(nullValue)
-    , ParamTypes(paramTypes)
+    , DestinationTypes(destinationTypes)
     , ParamSources(paramSources)
 {
     NCsvFormat::CsvSplitter splitter(HeaderRow, Delimeter);
     Header = static_cast<TVector<TString>>(splitter);
 }
 
 TCsvParser::TCsvParser(TVector<TString>&& header, const char delimeter, const std::optional<TString>& nullValue,
-                       const std::map<TString, TType>* paramTypes,
+                       const std::map<TString, TType>* destinationTypes,
                        const std::map<TString, TString>* paramSources) 
     : Header(std::move(header))
     , Delimeter(delimeter)
     , NullValue(nullValue)
-    , ParamTypes(paramTypes)
+    , DestinationTypes(destinationTypes)
     , ParamSources(paramSources)
 {
 }
 
-void TCsvParser::GetParams(TString&& data, TParamsBuilder& builder, const TParseMetadata& meta) const {
+void TCsvParser::BuildParams(TString& data, TParamsBuilder& builder, const TParseMetadata& meta) const {
     NCsvFormat::CsvSplitter splitter(data, Delimeter);
     auto headerIt = Header.begin();
     do {
@@ -374,8 +375,8 @@ void TCsvParser::GetParams(TString&& data, TParamsBuilder& builder, const TParse
         }
         TStringBuf token = Consume(splitter, meta, *headerIt);
         TString fullname = "$" + *headerIt;
-        auto paramIt = ParamTypes->find(fullname);
-        if (paramIt == ParamTypes->end()) {
+        auto paramIt = DestinationTypes->find(fullname);
+        if (paramIt == DestinationTypes->end()) {
             ++headerIt;
             continue;
         }
@@ -395,7 +396,7 @@ void TCsvParser::GetParams(TString&& data, TParamsBuilder& builder, const TParse
     }
 }
 
-void TCsvParser::GetValue(TString&& data, TValueBuilder& builder, const TType& type, const TParseMetadata& meta) const {
+void TCsvParser::BuildValue(TString& data, TValueBuilder& builder, const TType& type, const TParseMetadata& meta) const {
     NCsvFormat::CsvSplitter splitter(data, Delimeter);
     auto headerIt = Header.cbegin();
     std::map<TString, TStringBuf> fields;
@@ -431,18 +432,68 @@ void TCsvParser::GetValue(TString&& data, TValueBuilder& builder, const TType& t
     builder.EndStruct();
 }
 
-TType TCsvParser::GetColumnsType() const {
+TValue TCsvParser::BuildList(std::vector<TString>& lines, const TString& filename, std::optional<ui64> row) const {
+    std::vector<std::unique_ptr<TTypeParser>> columnTypeParsers;
+    columnTypeParsers.reserve(ResultColumnCount);
+    for (const TType* type : ResultLineTypesSorted) {
+        columnTypeParsers.push_back(std::make_unique<TTypeParser>(*type));
+    }
+    Ydb::Value listValue;
+    auto* listItems = listValue.mutable_items();
+    listItems->Reserve(lines.size());
+    for (auto& line : lines) {
+        std::vector<TStringBuf> fields;
+        NCsvFormat::CsvSplitter splitter(line, Delimeter);
+        TParseMetadata meta {row, filename};
+        auto headerIt = Header.cbegin();
+        auto skipIt = SkipBitMap.begin();
+        do {
+            if (headerIt == Header.cend()) { // SkipBitMap has same size as Header
+                throw FormatError(yexception() << "Header contains less fields than data. Header: \"" << HeaderRow << "\", data: \"" << line << "\"", meta);
+            }
+            TStringBuf nextField = Consume(splitter, meta, *headerIt);
+            if (!*skipIt) {
+                fields.emplace_back(nextField);
+            }
+            ++headerIt;
+            ++skipIt;
+        } while (splitter.Step());
+        auto* structItems = listItems->Add()->mutable_items();
+        structItems->Reserve(ResultColumnCount);
+        auto typeParserIt = columnTypeParsers.begin();
+        auto fieldIt = fields.begin();
+        auto nameIt = ResultLineNamesSorted.begin();
+        // fields size equals columnTypeParsers size, no need for second end check
+        for (; typeParserIt != columnTypeParsers.end(); ++typeParserIt, ++fieldIt, ++nameIt) {
+            *structItems->Add() = FieldToValue(*typeParserIt->get(), *fieldIt, NullValue, meta, **nameIt).GetProto();
+        }
+        if (row.has_value()) {
+            ++row.value();
+        }
+    }
+    return TValue(ResultListType.value(), std::move(listValue));
+}
+
+void TCsvParser::BuildLineType() {
+    SkipBitMap.reserve(Header.size());
+    ResultColumnCount = 0;
     TTypeBuilder builder;
     builder.BeginStruct();
     for (const auto& colName : Header) {
-        if (ParamTypes->find(colName) != ParamTypes->end()) {
-            builder.AddMember(colName, ParamTypes->at(colName));
+        auto findIt = DestinationTypes->find(colName);
+        if (findIt != DestinationTypes->end()) {
+            builder.AddMember(colName, findIt->second);
+            ResultLineTypesSorted.emplace_back(&findIt->second);
+            ResultLineNamesSorted.emplace_back(&colName);
+            SkipBitMap.push_back(false);
+            ++ResultColumnCount;
         } else {
-            builder.AddMember("__ydb_skip_column_name", TTypeBuilder().Build());
+            SkipBitMap.push_back(true);
         }
     }
     builder.EndStruct();
-    return builder.Build();
+    ResultLineType = builder.Build();
+    ResultListType = TTypeBuilder().List(ResultLineType.value()).Build();
 }
 
 }

ydb/public/lib/ydb_cli/common/csv_parser.h

@@ -25,23 +25,39 @@ class TCsvParser {
     ~TCsvParser() = default;
 
     TCsvParser(TString&& headerRow, const char delimeter, const std::optional<TString>& nullValue,
-               const std::map<TString, TType>* paramTypes = nullptr,
+               const std::map<TString, TType>* destinationTypes = nullptr,
                const std::map<TString, TString>* paramSources = nullptr);
     TCsvParser(TVector<TString>&& header, const char delimeter, const std::optional<TString>& nullValue,
-               const std::map<TString, TType>* paramTypes = nullptr,
+               const std::map<TString, TType>* destinationTypes = nullptr,
                const std::map<TString, TString>* paramSources = nullptr);
 
-    void GetParams(TString&& data, TParamsBuilder& builder, const TParseMetadata& meta) const;
-    void GetValue(TString&& data, TValueBuilder& builder, const TType& type, const TParseMetadata& meta) const;
-    TType GetColumnsType() const;
+    void BuildParams(TString& data, TParamsBuilder& builder, const TParseMetadata& meta) const;
+    void BuildValue(TString& data, TValueBuilder& builder, const TType& type, const TParseMetadata& meta) const;
+    TValue BuildList(std::vector<TString>& lines, const TString& filename,
+                     std::optional<ui64> row = std::nullopt) const;
+    void BuildLineType();
 
 private:
     TVector<TString> Header;
     TString HeaderRow;
     char Delimeter;
     std::optional<TString> NullValue;
-    const std::map<TString, TType>* ParamTypes;
+    // Types of destination table or query parameters
+    // Column name -> column type
+    const std::map<TString, TType>* DestinationTypes;
     const std::map<TString, TString>* ParamSources;
+    // Type of a single row in resulting TValue.
+    // Column order according to the header, though can have less elements than the Header
+    std::optional<TType> ResultLineType = std::nullopt;
+    std::optional<TType> ResultListType = std::nullopt;
+    // If a value is true (header column is absent in dstTypes), skip corresponding value in input csv row
+    std::vector<bool> SkipBitMap;
+    // Count of columns in each struct in resulting TValue
+    size_t ResultColumnCount;
+    // Types of columns in a single row in resulting TValue.
+    // Column order according to the header, though can have less elements than the Header
+    std::vector<const TType*> ResultLineTypesSorted;
+    std::vector<const TString*> ResultLineNamesSorted;
 };
 
 }