Skip to content

[kafka] Unify schema type validation for Kafka table mappings #4766

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wuchong merged 1 commit into from
Oct 23, 2026
+363 −73
Merged

[kafka] Unify schema type validation for Kafka table mappings #4766

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 37 additions & 39 deletions fluss-common/src/main/java/org/apache/fluss/metadata/KafkaTableSchema.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
import org.apache.fluss.config.TableConfig;
import org.apache.fluss.exception.InvalidTableException;
import org.apache.fluss.types.ArrayType;
import org.apache.fluss.types.BytesType;
import org.apache.fluss.types.DataField;
import org.apache.fluss.types.DataType;
import org.apache.fluss.types.DataTypes;
Expand All @@ -43,6 +42,8 @@
import java.util.Objects;
import java.util.Set;

import static org.apache.fluss.types.DataTypeChecks.equalsIgnoreFieldDescriptionsAndIds;

/**
* Immutable Kafka record mapping, shared by native DDL validation and Kafka request handling.
* Resolving a contract never changes the schema or materializes configuration defaults.
Expand All @@ -59,6 +60,18 @@ public final class KafkaTableSchema {
/** Maximum recursive depth below each top-level JSON business field. */
public static final int MAX_JSON_SCHEMA_NESTING_DEPTH = 64;

private static final DataType KEY_TYPE = DataTypes.BYTES();
private static final DataType HEADERS_TYPE =
DataTypes.MAP(DataTypes.STRING(), DataTypes.BYTES());
private static final RowType METADATA_TYPE =
DataTypes.ROW(
DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)),
DataTypes.FIELD("timestamp_type", DataTypes.INT()),
DataTypes.FIELD("partition", DataTypes.INT()));
private static final DataType RESCUE_TYPE = DataTypes.STRING();
private static final DataType BINARY_VALUE_TYPE = DataTypes.BYTES();
private static final DataType STRING_VALUE_TYPE = DataTypes.STRING();

private static final Set<String> RESERVED_NAMES =
Collections.unmodifiableSet(
new HashSet<>(
Expand Down Expand Up @@ -167,33 +180,14 @@ public static KafkaTableSchema resolve(TableDescriptor table) {
: null;

RowType rowType = table.getSchema().getRowType();
int keyPosition = rowType.getFieldIndex(KEY_COLUMN);
if (keyPosition >= 0 && !(rowType.getTypeAt(keyPosition) instanceof BytesType)) {
throw invalid("_kafka_key must be BYTES.");
}
int keyPosition = validateSystemColumnType(rowType, KEY_COLUMN, KEY_TYPE);
if (options.getOptional(ConfigOptions.TABLE_KAFKA_CONVERT_KEY_TYPE).isPresent()
&& keyPosition < 0) {
throw invalid("table.kafka.convert.key.type requires _kafka_key BYTES.");
}
int headersPosition = rowType.getFieldIndex(HEADERS_COLUMN);
if (headersPosition >= 0) {
DataType headers = rowType.getTypeAt(headersPosition);
if (!(headers instanceof MapType)
|| !(((MapType) headers).getKeyType() instanceof StringType)
|| !(((MapType) headers).getValueType() instanceof BytesType)) {
throw invalid("_kafka_headers must be MAP<STRING, BYTES>.");
}
}
int rescuePosition = rowType.getFieldIndex(RESCUE_COLUMN);
if (rescuePosition >= 0
&& (!(rowType.getTypeAt(rescuePosition) instanceof StringType)
|| !rowType.getTypeAt(rescuePosition).isNullable())) {
throw invalid("_kafka_rescue must be nullable STRING.");
}
int metadataPosition = rowType.getFieldIndex(METADATA_COLUMN);
if (metadataPosition >= 0) {
validateMetadataType(rowType.getTypeAt(metadataPosition));
}
int headersPosition = validateSystemColumnType(rowType, HEADERS_COLUMN, HEADERS_TYPE);
int rescuePosition = validateSystemColumnType(rowType, RESCUE_COLUMN, RESCUE_TYPE);
int metadataPosition = validateSystemColumnType(rowType, METADATA_COLUMN, METADATA_TYPE);

List<Integer> valuePositions = new ArrayList<>();
if (valueConversionLayout == KafkaValueConversionLayout.FLAT) {
Expand All @@ -215,9 +209,9 @@ public static KafkaTableSchema resolve(TableDescriptor table) {
DataType type = rowType.getTypeAt(valuePosition);
boolean valid =
valueConversionType == KafkaValueConversionType.BINARY
? type instanceof BytesType
? equalsIgnoreFieldDescriptionsAndIds(BINARY_VALUE_TYPE, type)
: valueConversionType == KafkaValueConversionType.STRING
? type instanceof StringType
? equalsIgnoreFieldDescriptionsAndIds(STRING_VALUE_TYPE, type)
: type instanceof RowType;
if (!valid) {
throw invalid(
Expand All @@ -229,6 +223,8 @@ public static KafkaTableSchema resolve(TableDescriptor table) {
: "ROW")
+ " for "
+ valueConversionType
+ ", but was "
+ type.asSummaryString()
+ ".");
}
for (String name : rowType.getFieldNames()) {
Expand Down Expand Up @@ -349,20 +345,22 @@ private static void validateNullable(DataType type, String path) {
}
}

private static void validateMetadataType(DataType type) {
String message =
"_kafka_metadata must be ROW<timestamp TIMESTAMP_LTZ(3), "
+ "timestamp_type INT, partition INT>.";
if (!(type instanceof RowType)) {
throw invalid(message);
}
RowType row = (RowType) type;
if (!row.getFieldNames().equals(Arrays.asList("timestamp", "timestamp_type", "partition"))
|| !row.getTypeAt(0).copy(true).equals(DataTypes.TIMESTAMP_LTZ(3))
|| !row.getTypeAt(1).copy(true).equals(DataTypes.INT())
|| !row.getTypeAt(2).copy(true).equals(DataTypes.INT())) {
throw invalid(message);
private static int validateSystemColumnType(
RowType rowType, String columnName, DataType expectedType) {
int position = rowType.getFieldIndex(columnName);
if (position >= 0) {
DataType type = rowType.getTypeAt(position);
if (!equalsIgnoreFieldDescriptionsAndIds(expectedType, type)) {
throw invalid(
columnName
+ " must be "
+ expectedType.asSummaryString()
+ ", but was "
+ type.asSummaryString()
+ ".");
}
}
return position;
}

/** Returns the unchanged physical schema, including declared nullability. */
Expand Down
70 changes: 70 additions & 0 deletions fluss-common/src/main/java/org/apache/fluss/types/DataTypeChecks.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,20 @@ public static boolean equalsWithFieldId(DataType original, DataType that) {
return that.accept(new DataTypeEqualsWithFieldId(original));
}

/**
* Checks whether two data types are equal, ignoring row field descriptions and ids.
*
* <p>Row fields are compared by name and type in the same order. The comparison applies
* recursively to rows, arrays, and maps, preserving nullability and type parameters such as
* precision, scale, and length.
*/
public static boolean equalsIgnoreFieldDescriptionsAndIds(DataType original, DataType that) {
if (original.isNullable() != that.isNullable()) {
return false;
}
return that.accept(new DataTypeEqualsIgnoreFieldDescriptionsAndIds(original));
}

/**
* Check if the given data type is supported for statistics collection. Uses a whitelist
* approach to ensure only types with meaningful statistical properties are included.
Expand Down Expand Up @@ -219,4 +233,60 @@ protected Boolean defaultMethod(DataType that) {
return original.equals(that);
}
}

private static class DataTypeEqualsIgnoreFieldDescriptionsAndIds
extends DataTypeDefaultVisitor<Boolean> {
private final DataType original;

private DataTypeEqualsIgnoreFieldDescriptionsAndIds(DataType original) {
this.original = original;
}

@Override
public Boolean visit(RowType that) {
if (!(original instanceof RowType)) {
return false;
}
List<DataField> originalFields = ((RowType) original).getFields();
List<DataField> thatFields = that.getFields();
if (originalFields.size() != thatFields.size()) {
return false;
}
for (int i = 0; i < originalFields.size(); i++) {
DataField originalField = originalFields.get(i);
DataField thatField = thatFields.get(i);
if (!originalField.getName().equals(thatField.getName())
|| !equalsIgnoreFieldDescriptionsAndIds(
originalField.getType(), thatField.getType())) {
return false;
}
}
return true;
}

@Override
public Boolean visit(ArrayType that) {
if (!(original instanceof ArrayType)) {
return false;
}
return equalsIgnoreFieldDescriptionsAndIds(
((ArrayType) original).getElementType(), that.getElementType());
}

@Override
public Boolean visit(MapType that) {
if (!(original instanceof MapType)) {
return false;
}
MapType originalMap = (MapType) original;
return equalsIgnoreFieldDescriptionsAndIds(originalMap.getKeyType(), that.getKeyType())
&& equalsIgnoreFieldDescriptionsAndIds(
originalMap.getValueType(), that.getValueType());
}

@Override
protected Boolean defaultMethod(DataType that) {
return original.equals(that);
}
}
}
54 changes: 39 additions & 15 deletions fluss-common/src/test/java/org/apache/fluss/metadata/KafkaTableSchemaTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,7 @@ void testFixedValueColumnWithOptionalSystemFields(String format) {
.build(),
format)
.build(),
"nullable fields");
"NOT NULL");
for (int mask = 0; mask < 16; mask++) {
Schema.Builder schema = Schema.newBuilder().column("_kafka_value", valueType);
if ((mask & 1) != 0) {
Expand Down Expand Up @@ -396,11 +396,10 @@ void testSystemColumnTypes() {
DataTypes.ARRAY(DataTypes.STRING()),
DataTypes.MAP(DataTypes.INT(), DataTypes.BYTES()),
DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))) {
assertInvalid(withSystem("_kafka_headers", invalid), "MAP<STRING, BYTES>");
assertInvalid(withSystem("_kafka_headers", invalid), "_kafka_headers must be MAP");
}
assertInvalid(withSystem("_kafka_rescue", DataTypes.INT()), "nullable STRING");
assertInvalid(
withSystem("_kafka_rescue", DataTypes.STRING().copy(false)), "nullable STRING");
assertInvalid(withSystem("_kafka_rescue", DataTypes.INT()), "_kafka_rescue must be STRING");
assertInvalid(withSystem("_kafka_rescue", DataTypes.STRING().copy(false)), "_kafka_rescue");
}

@Test
Expand All @@ -415,7 +414,7 @@ void testSystemFieldsRequireNullableTypes() {
withSystem(
"_kafka_headers",
DataTypes.MAP(DataTypes.STRING(), DataTypes.BYTES().copy(false))),
"_kafka_headers<value>");
"_kafka_headers");
assertInvalid(withSystem("_kafka_metadata", metadataType().copy(false)), "_kafka_metadata");
for (int i = 0; i < metadataType().getFieldCount(); i++) {
RowType.Builder metadata = RowType.builder();
Expand All @@ -424,9 +423,7 @@ void testSystemFieldsRequireNullableTypes() {
metadataType().getFields().get(j).getName(),
metadataType().getTypeAt(j).copy(i != j));
}
assertInvalid(
withSystem("_kafka_metadata", metadata.build()),
"_kafka_metadata." + metadataType().getFields().get(i).getName());
assertInvalid(withSystem("_kafka_metadata", metadata.build()), "_kafka_metadata");
}
}

Expand Down Expand Up @@ -493,6 +490,35 @@ void testOptionalBinaryKeyConversion() {
.hasMessageContaining("table.kafka.convert.key.type");
}

@Test
void testMetadataIgnoresFieldIdsAndComments() {
RowType metadata =
DataTypes.ROW(
DataTypes.FIELD(
"timestamp", DataTypes.TIMESTAMP_LTZ(3), "Record timestamp", 10),
DataTypes.FIELD("timestamp_type", DataTypes.INT(), "Timestamp type", 20),
DataTypes.FIELD("partition", DataTypes.INT(), "Requested partition", 30));
TableDescriptor table = withSystem("_kafka_metadata", metadata);

KafkaTableSchema resolved = KafkaTableSchema.resolve(table);

assertThat(resolved.rowType().getTypeAt(resolved.metadataPosition())).isEqualTo(metadata);
assertThat(resolved.rowType()).isEqualTo(table.getSchema().getRowType());
}

@ParameterizedTest
@ValueSource(ints = {0, 1, 2})
void testMetadataRejectsRenamedFields(int renamedField) {
RowType expected = metadataType();
RowType.Builder renamed = RowType.builder();
for (int i = 0; i < expected.getFieldCount(); i++) {
String name = expected.getFields().get(i).getName();
renamed.field(i == renamedField ? "renamed_" + name : name, expected.getTypeAt(i));
}
assertInvalid(
withSystem("_kafka_metadata", renamed.build()), "_kafka_metadata must be ROW");
}

@Test
void testMetadataRejectsInvalidStructureAndOffsetField() {
for (DataType invalid :
Expand Down Expand Up @@ -523,16 +549,14 @@ void testMetadataRejectsInvalidStructureAndOffsetField() {
DataTypes.FIELD("timestamp_type", DataTypes.INT()),
DataTypes.FIELD("partition", DataTypes.BIGINT())),
DataTypes.ROW(
DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)),
DataTypes.FIELD("partition", DataTypes.INT()),
DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)),
DataTypes.FIELD("timestamp_type", DataTypes.INT())),
DataTypes.ROW(
DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)),
DataTypes.FIELD("timestamp_type", DataTypes.INT()),
DataTypes.FIELD("bucket", DataTypes.INT())))) {
assertInvalid(
withSystem("_kafka_metadata", invalid),
"ROW<timestamp TIMESTAMP_LTZ(3), timestamp_type INT, partition INT>");
DataTypes.FIELD("partition", DataTypes.INT()),
DataTypes.FIELD("timestamp_type", DataTypes.INT())))) {
assertInvalid(withSystem("_kafka_metadata", invalid), "_kafka_metadata must be ROW");
}
}

Expand Down
Loading
Loading