Merge a03f6c6 into 31e0e46

Author: evanevanevanevannnn

ydb/library/yql/providers/s3/provider/yql_s3_datasink_type_ann.cpp

@@ -26,6 +26,23 @@ TExprNode::TListType GetPartitionKeys(const TExprNode::TPtr& partBy) {
 
     return {};
 }
+
+bool ValidateSchemaForOutput(const TStructExprType* schemaStructRowType, TExprContext& ctx) {
+    for (const TItemExprType* item : schemaStructRowType->GetItems()) {
+        const TTypeAnnotationNode* rowType = item->GetItemType();
+        if (rowType->GetKind() == ETypeAnnotationKind::Optional) {
+            rowType = rowType->Cast<TOptionalExprType>()->GetItemType();
+        }
+
+        if (rowType->GetKind() == ETypeAnnotationKind::Optional) {
+            ctx.AddError(TIssue(TStringBuilder() << "Double optional types are not supported for output (you have '"
+                << item->GetName() << " " << FormatType(item->GetItemType()) << "' field)"));
+            return false;
+        }
+    }
+    return true;
+}
+
 }
 
 namespace {
@@ -74,6 +91,10 @@ class TS3DataSinkTypeAnnotationTransformer : public TVisitorTransformerBase {
             return TStatus::Error;
         }
 
+        if (!ValidateSchemaForOutput(sourceType->Cast<TStructExprType>(), ctx)) {
+            return TStatus::Error;
+        }
+
         auto target = input->Child(TS3WriteObject::idx_Target);
         if (!TS3Target::Match(target)) {
             ctx.AddError(TIssue(ctx.GetPosition(target->Pos()), "Expected S3 target."));

ydb/tests/fq/s3/test_insert.py

@@ -535,3 +535,35 @@ def test_insert_without_format_error(self, kikimr, s3, client, unique_prefix):
         issues = str(client.describe_query(query_id).result.query.issue)
 
         assert "Missing format - please use WITH FORMAT when writing into S3" in issues, "Incorrect Issues: " + issues
+
+    @yq_all
+    @pytest.mark.parametrize("client", [{"folder_id": "my_folder"}], indirect=True)
+    def test_insert_type_validation(self, kikimr, s3, client, unique_prefix):
+        resource = boto3.resource(
+            "s3", endpoint_url=s3.s3_url, aws_access_key_id="key", aws_secret_access_key="secret_key"
+        )
+
+        bucket = resource.Bucket("fbucket")
+        bucket.create(ACL='public-read')
+        bucket.objects.all().delete()
+
+        s3_client = boto3.client(
+            "s3", endpoint_url=s3.s3_url, aws_access_key_id="key", aws_secret_access_key="secret_key"
+        )
+
+        kikimr.control_plane.wait_bootstrap(1)
+        storage_connection_name = unique_prefix + "fruitbucket"
+        client.create_storage_connection(storage_connection_name, "fbucket")
+
+        sql = f'''
+            INSERT INTO `{storage_connection_name}`.`insert/`
+            WITH
+            (
+                FORMAT="csv_with_names"
+            )
+            SELECT CAST(42 AS Int32??) as Weight;'''
+
+        query_id = client.create_query("simple", sql, type=fq.QueryContent.QueryType.ANALYTICS).result.query_id
+        client.wait_query_status(query_id, fq.QueryMeta.FAILED)
+
+        assert "Double optional types are not supported for output" in str(client.describe_query(query_id).result)