From 5dda221efc2c40f8fe9ae4e57ad94948d1d41e98 Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" Date: Sat, 1 Aug 2026 22:40:18 +0800 Subject: [PATCH 1/6] feat: propagate column semantic types into Arrow schema metadata for bulk write Attach GreptimeDB field metadata to the Arrow schema sent on the bulk (Flight) write path so the server can auto-create missing tables: - timestamp column: greptime:time_index=true, marked non-nullable - tag/field columns: greptime:semantic_type=tag|field - JSON columns: greptime:type=Json This matches the server-side contract introduced in GreptimeTeam/greptimedb-enterprise#845. Signed-off-by: Lei, HUANG --- .../java/io/greptime/models/ArrowHelper.java | 57 ++++++++++++++++++- .../io/greptime/models/ArrowHelperTest.java | 36 ++++++++++++ 2 files changed, 91 insertions(+), 2 deletions(-) diff --git a/ingester-protocol/src/main/java/io/greptime/models/ArrowHelper.java b/ingester-protocol/src/main/java/io/greptime/models/ArrowHelper.java index cbbef3f..c482787 100644 --- a/ingester-protocol/src/main/java/io/greptime/models/ArrowHelper.java +++ b/ingester-protocol/src/main/java/io/greptime/models/ArrowHelper.java @@ -22,8 +22,10 @@ import io.greptime.v1.Common; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.HashMap; import java.util.Iterator; import java.util.List; +import java.util.Map; import org.apache.arrow.vector.BigIntVector; import org.apache.arrow.vector.BitVector; import org.apache.arrow.vector.DateDayVector; @@ -61,6 +63,7 @@ import org.apache.arrow.vector.types.TimeUnit; import org.apache.arrow.vector.types.pojo.ArrowType; import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.FieldType; import org.apache.arrow.vector.types.pojo.Schema; /** @@ -68,6 +71,27 @@ */ public class ArrowHelper { + /** + * The Arrow field metadata key that marks a column as the time index. + * The value must be "true". Recognized by the GreptimeDB server when + * auto-creating tables for bulk insert requests. + */ + public static final String TIME_INDEX_METADATA_KEY = "greptime:time_index"; + + /** + * The Arrow field metadata key that carries the semantic type of a column. + * The value must be "tag" or "field". Recognized by the GreptimeDB server + * when auto-creating tables for bulk insert requests. + */ + public static final String SEMANTIC_TYPE_METADATA_KEY = "greptime:semantic_type"; + + /** + * The Arrow field metadata key that carries the GreptimeDB extended data type + * of a column, e.g. "Json". Recognized by the GreptimeDB server when + * auto-creating tables for bulk insert requests. + */ + public static final String DATA_TYPE_METADATA_KEY = "greptime:type"; + /** * Get the Arrow compression type from the context. * @@ -89,6 +113,13 @@ public static ArrowCompressionType getArrowCompressionType(Context ctx) { /** * Create an Arrow schema from a table schema. + *

+ * The semantic types of the columns are encoded as Arrow field metadata so that + * the GreptimeDB server can auto-create the table for bulk insert requests: + * the timestamp column is marked with `greptime:time_index=true` and is + * non-nullable, tag and field columns carry `greptime:semantic_type=tag` and + * `greptime:semantic_type=field` respectively, and JSON columns carry + * `greptime:type=Json`. * * @param tableSchema the table schema * @return the Arrow schema @@ -101,6 +132,7 @@ public static Schema createSchema(TableSchema tableSchema) { List fields = new ArrayList<>(columnCount); List columnNames = tableSchema.getColumnNames(); + List semanticTypes = tableSchema.getSemanticTypes(); List dataTypes = tableSchema.getDataTypes(); List dataTypeExtensions = tableSchema.getDataTypeExtensions(); @@ -108,8 +140,29 @@ public static Schema createSchema(TableSchema tableSchema) { String name = columnNames.get(i); ArrowType type = convertToArrowType(dataTypes.get(i), dataTypeExtensions.get(i)); - Field field = Field.nullable(name, type); - fields.add(field); + Map metadata = new HashMap<>(2); + boolean nullable = true; + switch (semanticTypes.get(i)) { + case TIMESTAMP: + metadata.put(TIME_INDEX_METADATA_KEY, "true"); + // The time index column is never null in GreptimeDB, and the server + // requires it to be non-nullable when auto-creating the table. + nullable = false; + break; + case TAG: + metadata.put(SEMANTIC_TYPE_METADATA_KEY, "tag"); + break; + case FIELD: + metadata.put(SEMANTIC_TYPE_METADATA_KEY, "field"); + break; + default: + throw new IllegalArgumentException("Unsupported semantic type: " + semanticTypes.get(i)); + } + if (dataTypes.get(i) == Common.ColumnDataType.JSON) { + metadata.put(DATA_TYPE_METADATA_KEY, "Json"); + } + + fields.add(new Field(name, new FieldType(nullable, type, null, metadata), null)); } return new Schema(fields); diff --git a/ingester-protocol/src/test/java/io/greptime/models/ArrowHelperTest.java b/ingester-protocol/src/test/java/io/greptime/models/ArrowHelperTest.java index eaae8bc..537ca28 100644 --- a/ingester-protocol/src/test/java/io/greptime/models/ArrowHelperTest.java +++ b/ingester-protocol/src/test/java/io/greptime/models/ArrowHelperTest.java @@ -20,6 +20,7 @@ import org.apache.arrow.vector.types.FloatingPointPrecision; import org.apache.arrow.vector.types.TimeUnit; import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; import org.apache.arrow.vector.types.pojo.Schema; import org.junit.Assert; import org.junit.Test; @@ -150,4 +151,39 @@ public void testCreateSchema() { new ArrowType.Binary().getTypeID(), schema.getFields().get(22).getType().getTypeID()); } + + @Test + public void testCreateSchemaWithSemanticTypeMetadata() { + TableSchema tableSchema = TableSchema.newBuilder("my_table") + .addTag("tag1", DataType.String) + .addTimestamp("ts", DataType.TimestampMillisecond) + .addField("field1", DataType.Float64) + .addField("field2", DataType.Json) + .build(); + + Schema schema = ArrowHelper.createSchema(tableSchema); + + Field tagField = schema.getFields().get(0); + Assert.assertTrue(tagField.isNullable()); + Assert.assertEquals("tag", tagField.getMetadata().get(ArrowHelper.SEMANTIC_TYPE_METADATA_KEY)); + Assert.assertFalse(tagField.getMetadata().containsKey(ArrowHelper.TIME_INDEX_METADATA_KEY)); + Assert.assertFalse(tagField.getMetadata().containsKey(ArrowHelper.DATA_TYPE_METADATA_KEY)); + + Field tsField = schema.getFields().get(1); + Assert.assertFalse(tsField.isNullable()); + Assert.assertEquals("true", tsField.getMetadata().get(ArrowHelper.TIME_INDEX_METADATA_KEY)); + Assert.assertFalse(tsField.getMetadata().containsKey(ArrowHelper.SEMANTIC_TYPE_METADATA_KEY)); + Assert.assertFalse(tsField.getMetadata().containsKey(ArrowHelper.DATA_TYPE_METADATA_KEY)); + + Field field1 = schema.getFields().get(2); + Assert.assertTrue(field1.isNullable()); + Assert.assertEquals("field", field1.getMetadata().get(ArrowHelper.SEMANTIC_TYPE_METADATA_KEY)); + Assert.assertFalse(field1.getMetadata().containsKey(ArrowHelper.TIME_INDEX_METADATA_KEY)); + Assert.assertFalse(field1.getMetadata().containsKey(ArrowHelper.DATA_TYPE_METADATA_KEY)); + + Field jsonField = schema.getFields().get(3); + Assert.assertTrue(jsonField.isNullable()); + Assert.assertEquals("field", jsonField.getMetadata().get(ArrowHelper.SEMANTIC_TYPE_METADATA_KEY)); + Assert.assertEquals("Json", jsonField.getMetadata().get(ArrowHelper.DATA_TYPE_METADATA_KEY)); + } } From 1c6f1c5a5172cf17516ee4ae4da3c6968788e985 Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" Date: Sat, 1 Aug 2026 23:48:26 +0800 Subject: [PATCH 2/6] refactor: use greptime:semantic_type=timestamp for the time index column Unify on a single metadata key: every column now carries greptime:semantic_type with value tag, field or timestamp, instead of marking the time index with a separate greptime:time_index=true key. Signed-off-by: Lei, HUANG --- .../java/io/greptime/models/ArrowHelper.java | 20 ++++++------------- .../io/greptime/models/ArrowHelperTest.java | 5 +---- 2 files changed, 7 insertions(+), 18 deletions(-) diff --git a/ingester-protocol/src/main/java/io/greptime/models/ArrowHelper.java b/ingester-protocol/src/main/java/io/greptime/models/ArrowHelper.java index c482787..5b0982a 100644 --- a/ingester-protocol/src/main/java/io/greptime/models/ArrowHelper.java +++ b/ingester-protocol/src/main/java/io/greptime/models/ArrowHelper.java @@ -71,17 +71,10 @@ */ public class ArrowHelper { - /** - * The Arrow field metadata key that marks a column as the time index. - * The value must be "true". Recognized by the GreptimeDB server when - * auto-creating tables for bulk insert requests. - */ - public static final String TIME_INDEX_METADATA_KEY = "greptime:time_index"; - /** * The Arrow field metadata key that carries the semantic type of a column. - * The value must be "tag" or "field". Recognized by the GreptimeDB server - * when auto-creating tables for bulk insert requests. + * The value must be "tag", "field" or "timestamp". Recognized by the + * GreptimeDB server when auto-creating tables for bulk insert requests. */ public static final String SEMANTIC_TYPE_METADATA_KEY = "greptime:semantic_type"; @@ -116,10 +109,9 @@ public static ArrowCompressionType getArrowCompressionType(Context ctx) { *

* The semantic types of the columns are encoded as Arrow field metadata so that * the GreptimeDB server can auto-create the table for bulk insert requests: - * the timestamp column is marked with `greptime:time_index=true` and is - * non-nullable, tag and field columns carry `greptime:semantic_type=tag` and - * `greptime:semantic_type=field` respectively, and JSON columns carry - * `greptime:type=Json`. + * each column carries {@code greptime:semantic_type} with value {@code tag}, + * {@code field} or {@code timestamp}, the timestamp (time index) column is + * non-nullable, and JSON columns additionally carry {@code greptime:type=Json}. * * @param tableSchema the table schema * @return the Arrow schema @@ -144,7 +136,7 @@ public static Schema createSchema(TableSchema tableSchema) { boolean nullable = true; switch (semanticTypes.get(i)) { case TIMESTAMP: - metadata.put(TIME_INDEX_METADATA_KEY, "true"); + metadata.put(SEMANTIC_TYPE_METADATA_KEY, "timestamp"); // The time index column is never null in GreptimeDB, and the server // requires it to be non-nullable when auto-creating the table. nullable = false; diff --git a/ingester-protocol/src/test/java/io/greptime/models/ArrowHelperTest.java b/ingester-protocol/src/test/java/io/greptime/models/ArrowHelperTest.java index 537ca28..7456e12 100644 --- a/ingester-protocol/src/test/java/io/greptime/models/ArrowHelperTest.java +++ b/ingester-protocol/src/test/java/io/greptime/models/ArrowHelperTest.java @@ -166,19 +166,16 @@ public void testCreateSchemaWithSemanticTypeMetadata() { Field tagField = schema.getFields().get(0); Assert.assertTrue(tagField.isNullable()); Assert.assertEquals("tag", tagField.getMetadata().get(ArrowHelper.SEMANTIC_TYPE_METADATA_KEY)); - Assert.assertFalse(tagField.getMetadata().containsKey(ArrowHelper.TIME_INDEX_METADATA_KEY)); Assert.assertFalse(tagField.getMetadata().containsKey(ArrowHelper.DATA_TYPE_METADATA_KEY)); Field tsField = schema.getFields().get(1); Assert.assertFalse(tsField.isNullable()); - Assert.assertEquals("true", tsField.getMetadata().get(ArrowHelper.TIME_INDEX_METADATA_KEY)); - Assert.assertFalse(tsField.getMetadata().containsKey(ArrowHelper.SEMANTIC_TYPE_METADATA_KEY)); + Assert.assertEquals("timestamp", tsField.getMetadata().get(ArrowHelper.SEMANTIC_TYPE_METADATA_KEY)); Assert.assertFalse(tsField.getMetadata().containsKey(ArrowHelper.DATA_TYPE_METADATA_KEY)); Field field1 = schema.getFields().get(2); Assert.assertTrue(field1.isNullable()); Assert.assertEquals("field", field1.getMetadata().get(ArrowHelper.SEMANTIC_TYPE_METADATA_KEY)); - Assert.assertFalse(field1.getMetadata().containsKey(ArrowHelper.TIME_INDEX_METADATA_KEY)); Assert.assertFalse(field1.getMetadata().containsKey(ArrowHelper.DATA_TYPE_METADATA_KEY)); Field jsonField = schema.getFields().get(3); From e5632ca9c92d5e35b1604631c199b01d957c2483 Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" Date: Wed, 5 Aug 2026 21:27:54 +0800 Subject: [PATCH 3/6] feat: add typed auto-create-table request hint Signed-off-by: Lei, HUANG --- .../src/main/java/io/greptime/rpc/Context.java | 13 +++++++++++++ .../src/test/java/io/greptime/rpc/ContextTest.java | 7 +++++++ 2 files changed, 20 insertions(+) diff --git a/ingester-rpc/src/main/java/io/greptime/rpc/Context.java b/ingester-rpc/src/main/java/io/greptime/rpc/Context.java index e4e74d8..7c007e1 100644 --- a/ingester-rpc/src/main/java/io/greptime/rpc/Context.java +++ b/ingester-rpc/src/main/java/io/greptime/rpc/Context.java @@ -28,6 +28,8 @@ @SuppressWarnings({"unchecked"}) public class Context { + private static final String AUTO_CREATE_TABLE_HINT = "auto_create_table"; + private final Map ctx = new HashMap<>(); private Compression compression = Compression.None; @@ -109,6 +111,17 @@ public Context withHint(String key, String value) { return this; } + /** + * Sets whether this request allows the server to auto-create a missing table. + * The server-side global auto-create-table option must also be enabled. + * + * @param enabled whether this request allows automatic table creation + * @return this {@link Context} + */ + public Context withAutoCreateTable(boolean enabled) { + return withHint(AUTO_CREATE_TABLE_HINT, Boolean.toString(enabled)); + } + /** * Gets the hints from the context. * diff --git a/ingester-rpc/src/test/java/io/greptime/rpc/ContextTest.java b/ingester-rpc/src/test/java/io/greptime/rpc/ContextTest.java index 2113ed2..96701e4 100644 --- a/ingester-rpc/src/test/java/io/greptime/rpc/ContextTest.java +++ b/ingester-rpc/src/test/java/io/greptime/rpc/ContextTest.java @@ -51,6 +51,13 @@ public void withHintTest() { Assert.assertEquals("key=value,key2=value2", context.getHints()); } + @Test + public void withAutoCreateTableTest() { + Context context = Context.newDefault().withAutoCreateTable(false); + + Assert.assertEquals("auto_create_table=false", context.getHints()); + } + @Test public void getShouldReturnNullForNonExistingKeyTest() { Context context = Context.newDefault(); From eb839a33d73ffbf056e7c1e63aeafc4e3a9ccb8c Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" Date: Mon, 17 Aug 2026 22:37:15 +0800 Subject: [PATCH 4/6] style: pre-size schema metadata map for JSON columns Signed-off-by: Lei, HUANG --- .../src/main/java/io/greptime/models/ArrowHelper.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ingester-protocol/src/main/java/io/greptime/models/ArrowHelper.java b/ingester-protocol/src/main/java/io/greptime/models/ArrowHelper.java index 5b0982a..86560f3 100644 --- a/ingester-protocol/src/main/java/io/greptime/models/ArrowHelper.java +++ b/ingester-protocol/src/main/java/io/greptime/models/ArrowHelper.java @@ -132,7 +132,8 @@ public static Schema createSchema(TableSchema tableSchema) { String name = columnNames.get(i); ArrowType type = convertToArrowType(dataTypes.get(i), dataTypeExtensions.get(i)); - Map metadata = new HashMap<>(2); + // JSON columns carry a second metadata entry (greptime:type=Json) + Map metadata = new HashMap<>(4); boolean nullable = true; switch (semanticTypes.get(i)) { case TIMESTAMP: From eeaab2d98770df2cff2f4cc3d33daf0fa05a3795 Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" Date: Mon, 17 Aug 2026 22:39:03 +0800 Subject: [PATCH 5/6] docs: document that withHint appends instead of overwriting Signed-off-by: Lei, HUANG --- ingester-rpc/src/main/java/io/greptime/rpc/Context.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/ingester-rpc/src/main/java/io/greptime/rpc/Context.java b/ingester-rpc/src/main/java/io/greptime/rpc/Context.java index 7c007e1..efd9e5c 100644 --- a/ingester-rpc/src/main/java/io/greptime/rpc/Context.java +++ b/ingester-rpc/src/main/java/io/greptime/rpc/Context.java @@ -92,6 +92,10 @@ public Context with(String key, Object value) { /** * Adds a hint to the context. + *

+ * Repeated calls append entries rather than overwriting: calling this twice + * with the same key produces a hint string like {@code key=value1,key=value2}, + * and which value takes effect is up to server-side parsing. * * @param key the key * @param value the value From 4d6e70cf672cf1433625aff1acc30aec0dea01ba Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" Date: Mon, 17 Aug 2026 22:39:03 +0800 Subject: [PATCH 6/6] docs: clarify withAutoCreateTable scope and server option asymmetry Signed-off-by: Lei, HUANG --- .../src/main/java/io/greptime/rpc/Context.java | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/ingester-rpc/src/main/java/io/greptime/rpc/Context.java b/ingester-rpc/src/main/java/io/greptime/rpc/Context.java index efd9e5c..f9c4cb7 100644 --- a/ingester-rpc/src/main/java/io/greptime/rpc/Context.java +++ b/ingester-rpc/src/main/java/io/greptime/rpc/Context.java @@ -117,7 +117,19 @@ public Context withHint(String key, String value) { /** * Sets whether this request allows the server to auto-create a missing table. - * The server-side global auto-create-table option must also be enabled. + *

+ * The hint travels with any write that uses this {@link Context}: it is sent + * on bulk writes (Flight) as well as on row-based writes, where it is carried + * as a gRPC header by {@code ContextToHeadersInterceptor}. + *

+ * Note the asymmetry: {@code false} always disables auto-creation for this + * request, while {@code true} cannot override a server whose global + * auto-create-table option is disabled. + *

+ * Like {@link #withHint}, repeated calls append instead of overwriting, so + * calling this twice on a reused {@link Context} yields duplicate + * {@code auto_create_table} hints. Use a fresh {@link Context} (or + * {@link #remove(String)} on the hints key) to change the value. * * @param enabled whether this request allows automatic table creation * @return this {@link Context}