Merge pull request #446 from Tmonster/disable_null_inserts_into_null_columns

Disable inserting NULL into columns marked NOT NULL

Author: Tmonster

scripts/data_generators/tests/test_not_null/__init__.py

@@ -0,0 +1,8 @@
+from scripts.data_generators.tests.base import IcebergTest
+import pathlib
+
+@IcebergTest.register()
+class Test(IcebergTest):
+    def __init__(self):
+        path = pathlib.PurePath(__file__)
+        super().__init__(path.parent.name)

scripts/data_generators/tests/test_not_null/q00.sql

@@ -0,0 +1,11 @@
+CREATE OR REPLACE TABLE default.test_not_null (
+    id INT,
+    name STRING NOT NULL,
+    address STRUCT<
+        street: STRING NOT NULL,
+        city: STRING NOT NULL,
+        zip: STRING NOT NULL
+    >,
+    phone_numbers ARRAY<STRING>,
+    metadata MAP<STRING, STRING>
+);

scripts/data_generators/tests/test_not_null/q01.sql

@@ -0,0 +1,7 @@
+INSERT INTO default.test_not_null VALUES (
+    1,
+    'Alice',
+    NAMED_STRUCT('street', '123 Main St', 'city', 'Metropolis', 'zip', '12345'),
+    ARRAY('123-456-7890', '987-654-3210'),
+    MAP('age', '30', 'membership', 'gold')
+);

src/storage/create_table/iceberg_create_table_request.cpp

@@ -2,6 +2,7 @@
 #include "storage/table_create/iceberg_create_table_request.hpp"
 #include "storage/irc_table_set.hpp"
 #include "storage/iceberg_table_information.hpp"
+#include "duckdb/parser/constraints/not_null_constraint.hpp"
 #include "utils/iceberg_type.hpp"
 #include "duckdb/common/enums/catalog_type.hpp"
 #include "duckdb/catalog/catalog_entry/copy_function_catalog_entry.hpp"
@@ -105,9 +106,23 @@ shared_ptr<IcebergTableSchema> IcebergCreateTableRequest::CreateIcebergSchema(co
 		return field_id++;
 	};
 
+	auto &constraints = table_entry->GetConstraints();
 	for (auto column = column_iterator.begin(); column != column_iterator.end(); ++column) {
 		auto name = (*column).Name();
+		// check if there is a not null constraint
 		bool required = false;
+		if (!constraints.empty()) {
+			for (auto &constraint : constraints) {
+				if (constraint->type != ConstraintType::NOT_NULL) {
+					continue;
+				}
+				auto &not_null_constraint = constraint->Cast<NotNullConstraint>();
+				if (not_null_constraint.index.IsValid() && not_null_constraint.index.index == column.pos) {
+					required = true;
+				}
+			}
+		}
+
 		auto logical_type = (*column).GetType();
 		idx_t first_id = next_field_id();
 		rest_api_objects::Type type;

src/storage/iceberg_insert.cpp

@@ -3,6 +3,7 @@
 #include "storage/irc_transaction.hpp"
 #include "storage/irc_table_entry.hpp"
 #include "storage/iceberg_table_information.hpp"
+#include "metadata/iceberg_column_definition.hpp"
 
 #include "iceberg_multi_file_list.hpp"
 
@@ -149,7 +150,26 @@ static IcebergColumnStats ParseColumnStats(const vector<Value> col_stats) {
 	return column_stats;
 }
 
-static void AddWrittenFiles(IcebergInsertGlobalState &global_state, DataChunk &chunk, optional_idx partition_id) {
+static void AddToColDefMap(case_insensitive_map_t<optional_ptr<IcebergColumnDefinition>> &name_to_coldef,
+                           string col_name_prefix, optional_ptr<IcebergColumnDefinition> column_def) {
+	string column_name = column_def->name;
+	if (!col_name_prefix.empty()) {
+		column_name = col_name_prefix + "." + column_def->name;
+	}
+	if (column_def->IsIcebergPrimitiveType()) {
+		name_to_coldef.emplace(column_name, column_def.get());
+	} else {
+		for (auto &child : column_def->children) {
+			AddToColDefMap(name_to_coldef, column_name, child.get());
+		}
+	}
+}
+
+static void AddWrittenFiles(IcebergInsertGlobalState &global_state, DataChunk &chunk,
+                            optional_ptr<TableCatalogEntry> table) {
+	D_ASSERT(table);
+	auto &ic_table = table->Cast<ICTableEntry>();
+	auto partition_id = ic_table.table_info.table_metadata.default_spec_id;
 	for (idx_t r = 0; r < chunk.size(); r++) {
 		IcebergManifestEntry data_file;
 		data_file.file_path = chunk.GetValue(0, r).GetValue<string>();
@@ -159,8 +179,8 @@ static void AddWrittenFiles(IcebergInsertGlobalState &global_state, DataChunk &c
 		data_file.status = IcebergManifestEntryStatusType::ADDED;
 		data_file.file_format = "parquet";
 
-		if (partition_id.IsValid()) {
-			data_file.partition_spec_id = static_cast<int32_t>(partition_id.GetIndex());
+		if (partition_id) {
+			data_file.partition_spec_id = static_cast<int32_t>(partition_id);
 		} else {
 			data_file.partition_spec_id = 0;
 		}
@@ -171,12 +191,27 @@ static void AddWrittenFiles(IcebergInsertGlobalState &global_state, DataChunk &c
 
 		global_state.insert_count += data_file.record_count;
 
+		auto table_current_schema_id = ic_table.table_info.table_metadata.current_schema_id;
+		auto ic_schema = ic_table.table_info.table_metadata.schemas[table_current_schema_id];
+
+		case_insensitive_map_t<optional_ptr<IcebergColumnDefinition>> column_info;
+		for (auto &column : ic_schema->columns) {
+			AddToColDefMap(column_info, "", column.get());
+		}
+
 		for (idx_t col_idx = 0; col_idx < map_children.size(); col_idx++) {
 			auto &struct_children = StructValue::GetChildren(map_children[col_idx]);
 			auto &col_name = StringValue::Get(struct_children[0]);
 			auto &col_stats = MapValue::GetChildren(struct_children[1]);
 			auto column_names = ParseQuotedList(col_name, '.');
 			auto stats = ParseColumnStats(col_stats);
+			auto normalized_col_name = StringUtil::Join(column_names, ".");
+
+			auto ic_column_info = column_info.find(normalized_col_name);
+			D_ASSERT(ic_column_info != column_info.end());
+			if (ic_column_info->second->required && stats.has_null_count && stats.null_count > 0) {
+				throw ConstraintException("NOT NULL constraint failed: %s.%s", table->name, normalized_col_name);
+			}
 
 			//! TODO: convert 'stats' into 'data_file.lower_bounds', upper_bounds, value_counts, null_value_counts,
 			//! nan_value_counts ...
@@ -205,7 +240,7 @@ SinkResultType IcebergInsert::Sink(ExecutionContext &context, DataChunk &chunk,
 	auto &global_state = input.global_state.Cast<IcebergInsertGlobalState>();
 
 	// TODO: pass through the partition id?
-	AddWrittenFiles(global_state, chunk, {});
+	AddWrittenFiles(global_state, chunk, table);
 
 	return SinkResultType::NEED_MORE_INPUT;
 }

test/sql/local/irc/create/test_create_table_not_null.test

@@ -0,0 +1,61 @@
+# name: test/sql/local/irc/create/test_create_table_not_null.test
+# description: test create table
+# group: [create]
+
+require-env ICEBERG_SERVER_AVAILABLE
+
+require avro
+
+require parquet
+
+require iceberg
+
+require httpfs
+
+# Do not ignore 'HTTP' error messages!
+set ignore_error_messages
+
+statement ok
+CALL enable_logging('HTTP');
+
+statement ok
+set logging_level='debug'
+
+statement ok
+CREATE SECRET (
+    TYPE S3,
+    KEY_ID 'admin',
+    SECRET 'password',
+    ENDPOINT '127.0.0.1:9000',
+    URL_STYLE 'path',
+    USE_SSL 0
+);
+
+
+statement ok
+ATTACH '' AS my_datalake (
+    TYPE ICEBERG,
+    CLIENT_ID 'admin',
+    CLIENT_SECRET 'password',
+    ENDPOINT 'http://127.0.0.1:8181'
+);
+
+statement ok
+use my_datalake.default;
+
+statement ok
+drop table if exists table_with_null_reqs;
+
+statement ok
+create table table_with_null_reqs (
+	a varchar,
+	b int NOT NULL
+);
+
+statement error
+insert into table_with_null_reqs values ('value', NULL);
+----
+<REGEX>:.*NOT NULL constraint failed.*table_with_null_reqs.b.*
+
+statement ok
+insert into table_with_null_reqs values ('value', 5);

test/sql/local/irc/insert/test_insert_into_null_columns.test

@@ -0,0 +1,74 @@
+# name: test/sql/local/irc/insert/test_insert_into_null_columns.test
+# group: [insert]
+
+require-env ICEBERG_SERVER_AVAILABLE
+
+require avro
+
+require parquet
+
+require iceberg
+
+require httpfs
+
+# Do not ignore 'HTTP' error messages!
+set ignore_error_messages
+
+statement ok
+CALL enable_logging('HTTP');
+
+statement ok
+set logging_level='debug'
+
+statement ok
+CREATE SECRET (
+    TYPE S3,
+    KEY_ID 'admin',
+    SECRET 'password',
+    ENDPOINT '127.0.0.1:9000',
+    URL_STYLE 'path',
+    USE_SSL 0
+);
+
+
+statement ok
+ATTACH '' AS my_datalake (
+    TYPE ICEBERG,
+    CLIENT_ID 'admin',
+    CLIENT_SECRET 'password',
+    ENDPOINT 'http://127.0.0.1:8181'
+);
+
+# first test that we can insert nested types
+statement ok
+insert into my_datalake.default.test_not_null VALUES (
+	1,
+	'NULL nested type',
+	{'street': 'duck street', 'city': 'Metropolis', 'zip': '12345'},
+	['123-456-7890', '987-654-3210']::VARCHAR[],
+	MAP {'age': '30', 'membership': 'gold'}
+);
+
+# cannot insert null into a nested type that is required
+statement error
+insert into my_datalake.default.test_not_null VALUES (
+	1,
+	'NULL nested type',
+	{'street': NULL, 'city': 'Metropolis', 'zip': '12345'},
+	['123-456-7890', '987-654-3210']::VARCHAR[],
+	MAP {'age': '30', 'membership': 'gold'}
+);
+----
+<REGEX>:.*Constraint Error.*NOT NULL.*
+
+# cannot insert NULL into primitive type marked NOT NULL
+statement error
+insert into my_datalake.default.test_not_null VALUES (
+	1,
+	NULL,
+	{'street': 'duck street', 'city': 'swan city', 'zip': '12345'},
+	['123-456-7890', '987-654-3210']::VARCHAR[],
+	MAP {'age': '30', 'membership': 'gold'}
+);
+----
+<REGEX>:.*Constraint Error.*NOT NULL.*