laskoviymishka commented on code in PR #17653:
URL: https://github.com/apache/iceberg/pull/17653#discussion_r3986975601
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java:
##########
@@ -179,6 +181,12 @@ private static ConfigDef newConfigDef() {
false,
Importance.MEDIUM,
"Set to true to add any missing record fields to the table schema,
false otherwise");
+ configDef.define(
+ TABLES_REPLACE_NULL_WITH_DEFAULT_PROP,
+ ConfigDef.Type.BOOLEAN,
+ true,
Review Comment:
Worth settling the default before this ships, since the key becomes a public
contract. Defaulting to `true` keeps today's behavior — an explicit null
silently written as the column default — for everyone who doesn't read the
release notes and opt out, which is the behavior this PR frames as the bug. A
`false` default with a one-line migration note would land users on the correct
semantics by default; the catch is the required-column failure above, so the
two decisions are linked.
I lean toward keeping `true` for back-compat and documenting loudly, but I'd
like this to be a deliberate call rather than a default-by-omission. wdyt?
##########
docs/docs/kafka-connect.md:
##########
@@ -71,6 +71,7 @@ for exactly-once semantics. This requires Kafka 2.5 or later.
| iceberg.tables.evolve-schema-enabled | Set to `true` to add any
missing record fields to the table schema, default is `false`
|
| iceberg.tables.schema-force-optional | Set to `true` to set columns as
optional during table create and evolution, default is `false` to respect
schema |
| iceberg.tables.schema-case-insensitive | Set to `true` to look up table
columns by case-insensitive name, default is `false` for case-sensitive
|
+| iceberg.tables.replace-null-with-default | Set to `false` to preserve
explicit null values instead of replacing them with the record schema default
value, default is `true` |
Review Comment:
One thing missing from the docs: with `replace-null-with-default=false`, an
explicit null flowing into a required Iceberg column doesn't get rescued by the
table's write-default (that only fills fields omitted from the write, not
explicit nulls), so the null reaches the Parquet/ORC writer and fails with an
opaque internal error. That's exactly the common auto-created-table + Debezium
CDC case.
I'd add a note that required columns can't take explicit nulls under this
setting, and point at `iceberg.tables.schema-force-optional=true` (or
pre-altering the columns) as the mitigation. A `LOG.warn` when a null lands on
a required field would make the failure diagnosable too, though that part's
optional.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java:
##########
@@ -259,7 +259,7 @@ private GenericRecord convertToStruct(
hasSchemaUpdates = true;
}
}
- Object recordFieldValue = struct.get(recordField);
+ Object recordFieldValue = fieldValue(struct, recordField);
Review Comment:
I think there's a subtle issue here once `replace-null-with-default=false`.
Before this change, `struct.get(recordField)` returned the Connect schema
default for a stored-null field, so `recordFieldValue` was non-null and the
null-gated branch on the next line never fired. Now `getWithoutDefault` hands
back the real null, so an explicit null on a field that carries a schema
default drops straight into `evolveSchemaFromConnectSchema` whenever a
schema-update consumer is present. Scalars no-op, but a nullable struct-typed
field recurses and can emit addColumn/makeOptional off a value that isn't
actually there — spurious, and those schema mutations aren't reversible.
None of the current tests exercise this: `testReplaceNullWithDefault`
doesn't pass a `SchemaUpdate.Consumer`, so line 263 is never hit with the new
path. I'd add a case that enables both `evolve-schema-enabled` and
`replace-null-with-default=false` with a nullable struct column and asserts no
evolution events fire. Either the branch needs a guard for the preserve-null
case, or we confirm the recursion is genuinely a no-op and lock it down with
that test. wdyt?
##########
kafka-connect/kafka-connect-transforms/src/main/java/org/apache/iceberg/connect/transforms/CopyValue.java:
##########
@@ -90,9 +90,10 @@ private R applyWithSchema(R record) {
Struct updatedValue = new Struct(updatedSchema);
for (Field field : value.schema().fields()) {
- updatedValue.put(field.name(), value.get(field));
+ // getWithoutDefault so an explicit null is not replaced by the schema
default value
+ updatedValue.put(field.name(), value.getWithoutDefault(field.name()));
Review Comment:
The SMT switch is unconditional — unlike the sink path there's no
`replace-null-with-default` knob here, so anyone using these public transforms
(CopyValue, DebeziumTransform, KafkaMetadataTransform) downstream of a
non-Iceberg consumer gets a silent behavioral change: a stored null that used
to surface as the schema default now propagates as raw null, with no way to
restore the old behavior.
That's defensible as the correct semantics, but it's a public-API change
with no migration note. I'd at least call out in kafka-connect.md that the
bundled SMTs always preserve explicit nulls regardless of the sink option.
While we're here, `MongoDebeziumTransform` still uses `get(...)` and wasn't
updated — worth the same treatment so the SMTs stay consistent.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java:
##########
@@ -179,6 +181,12 @@ private static ConfigDef newConfigDef() {
false,
Importance.MEDIUM,
"Set to true to add any missing record fields to the table schema,
false otherwise");
+ configDef.define(
+ TABLES_REPLACE_NULL_WITH_DEFAULT_PROP,
+ ConfigDef.Type.BOOLEAN,
+ true,
+ Importance.MEDIUM,
+ "Set to false to preserve explicit null values instead of replacing
them with the record schema default value, true otherwise");
Review Comment:
Every other boolean config here leads with the `true` case; this one leads
with `false`, which reads as a double negative. Minor, but for consistency:
```suggestion
"Set to true to replace null struct field values with the record
schema default value, false to preserve explicit nulls (default is true)");
```
##########
docs/docs/kafka-connect.md:
##########
@@ -94,6 +95,11 @@ If `iceberg.tables.dynamic-enabled` is `false` (the default)
then you must speci
`iceberg.tables.dynamic-enabled` is `true` then you must specify
`iceberg.tables.route-field` which will
contain the name of the table.
+When `iceberg.tables.replace-null-with-default` is set to `false`, a record
whose route field is
Review Comment:
Two small things on this paragraph. It only describes the route-field side
effect, so a reader scanning the prose (not the table) might think the option
is purely about routing — I'd open with a sentence on the general purpose
(preserving explicit nulls end-to-end) before the routing caveat.
And on the JSON converter note: worth saying its `replace.null.with.default`
has defaulted to `true` since Kafka 3.6.0, so the substitution already happens
at the converter layer by default — setting only the sink option leaves the new
behavior silently ineffective.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java:
##########
@@ -155,6 +159,51 @@ public void testDynamicRoute() {
assertThat(writerResult.tableReference().identifier()).isEqualTo(TABLE_IDENTIFIER);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testDynamicRouteReplaceNullWithDefault(boolean
replaceNullWithDefault) {
Review Comment:
This covers the dynamic route path, but `extractRouteValue` is also called
from `routeRecordStatically`, and that path isn't tested with the new config
plumbed through. I'd add a static-route counterpart so we know the config
actually reaches both.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java:
##########
@@ -362,6 +362,18 @@ private void logMismatchedType(
"Record schema of type {} does not match table of type {}",
recordSchemaType, tableType);
}
+ /**
+ * Reads a struct field value. {@link Struct#get(Field)} substitutes the
schema default value when
+ * the stored value is null, which turns an explicit null into the default;
whether that
+ * substitution happens is controlled by the {@code
iceberg.tables.replace-null-with-default}
+ * setting.
+ */
+ private Object fieldValue(Struct struct, Field field) {
+ return config.replaceNullWithDefault()
Review Comment:
Non-blocker: this is the same abstraction as
`RecordUtils.fieldValueFromStruct` under a different name — I'd match them
(`fieldValue` → `fieldValueFromStruct`) so the two read as the pair they are.
Also `config.replaceNullWithDefault()` does a map lookup per field on every
struct and variant recursion; the value is fixed for the converter's lifetime,
so caching it in a `final boolean` in the constructor would keep the hot path
clean.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestRecordUtils.java:
##########
@@ -61,17 +74,37 @@ public void testExtractFromRecordValueStructNull() {
Schema valSchema = SchemaBuilder.struct().field("key",
Schema.INT64_SCHEMA).build();
Struct val = new Struct(valSchema).put("key", 123L);
- Object result = RecordUtils.extractFromRecordValue(val, "");
+ Object result = RecordUtils.extractFromRecordValue(val, "", config);
assertThat(result).isNull();
- result = RecordUtils.extractFromRecordValue(val, "xkey");
+ result = RecordUtils.extractFromRecordValue(val, "xkey", config);
assertThat(result).isNull();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testExtractFromRecordValueStructReplaceNullWithDefault(
Review Comment:
This only exercises a flat single-field struct, but `valueFromStruct` walks
dotted paths (`data.id.key`). I'd add a two-level case where the leaf field has
a schema default and a stored null, so the preserve-null behavior is covered
through the nested walk and not just the top level.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
