This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 55447d1d7 [core] support to show comments for non-physical columns
(#2529)
55447d1d7 is described below
commit 55447d1d7982a4d771b15b0d81abb30de93c024c
Author: Aitozi <[email protected]>
AuthorDate: Tue Dec 19 10:41:12 2023 +0800
[core] support to show comments for non-physical columns (#2529)
---
paimon-flink/paimon-flink-common/pom.xml | 9 +++
.../org/apache/paimon/flink/DataCatalogTable.java | 18 +++++-
.../java/org/apache/paimon/flink/FlinkCatalog.java | 50 +++++++--------
.../flink/utils/FlinkCatalogPropertiesUtil.java | 30 +++++++++
.../apache/paimon/flink/CatalogTableITCase.java | 14 ++++
.../org/apache/paimon/flink/FlinkCatalogTest.java | 74 ++++++++++++++++------
.../apache/paimon/flink/ReadWriteTableITCase.java | 8 +--
7 files changed, 151 insertions(+), 52 deletions(-)
diff --git a/paimon-flink/paimon-flink-common/pom.xml
b/paimon-flink/paimon-flink-common/pom.xml
index c5a7d32a7..3c49ecce5 100644
--- a/paimon-flink/paimon-flink-common/pom.xml
+++ b/paimon-flink/paimon-flink-common/pom.xml
@@ -210,6 +210,15 @@ under the License.
<version>${flink.version}</version>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-common</artifactId>
+ <version>${flink.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
<build>
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/DataCatalogTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/DataCatalogTable.java
index 37d1f9fa3..019d7bd68 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/DataCatalogTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/DataCatalogTable.java
@@ -40,15 +40,18 @@ import java.util.stream.Collectors;
public class DataCatalogTable extends CatalogTableImpl {
private final Table table;
+ private final Map<String, String> nonPhysicalColumnComments;
public DataCatalogTable(
Table table,
TableSchema tableSchema,
List<String> partitionKeys,
Map<String, String> properties,
- String comment) {
+ String comment,
+ Map<String, String> nonPhysicalColumnComments) {
super(tableSchema, partitionKeys, properties, comment);
this.table = table;
+ this.nonPhysicalColumnComments = nonPhysicalColumnComments;
}
public Table table() {
@@ -97,6 +100,8 @@ public class DataCatalogTable extends CatalogTableImpl {
String colName = column.getName();
if (comments.containsKey(colName)) {
builder.withComment(comments.get(colName));
+ } else if
(nonPhysicalColumnComments.containsKey(colName)) {
+
builder.withComment(nonPhysicalColumnComments.get(colName));
}
});
@@ -122,11 +127,18 @@ public class DataCatalogTable extends CatalogTableImpl {
getSchema().copy(),
new ArrayList<>(getPartitionKeys()),
new HashMap<>(getOptions()),
- getComment());
+ getComment(),
+ nonPhysicalColumnComments);
}
@Override
public CatalogTable copy(Map<String, String> options) {
- return new DataCatalogTable(table, getSchema(), getPartitionKeys(),
options, getComment());
+ return new DataCatalogTable(
+ table,
+ getSchema(),
+ getPartitionKeys(),
+ options,
+ getComment(),
+ nonPhysicalColumnComments);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
index c854be93a..e8a0f9635 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
@@ -38,9 +38,7 @@ import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.RowDataPartitionComputer;
import org.apache.paimon.utils.StringUtils;
-import org.apache.flink.table.api.TableColumn;
import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.api.WatermarkSpec;
import org.apache.flink.table.catalog.AbstractCatalog;
import org.apache.flink.table.catalog.CatalogBaseTable;
import org.apache.flink.table.catalog.CatalogDatabase;
@@ -52,6 +50,8 @@ import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.CatalogTableImpl;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.ResolvedCatalogBaseTable;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TableChange;
import org.apache.flink.table.catalog.TableChange.AddColumn;
import org.apache.flink.table.catalog.TableChange.AddWatermark;
@@ -67,6 +67,7 @@ import
org.apache.flink.table.catalog.TableChange.ModifyPhysicalColumnType;
import org.apache.flink.table.catalog.TableChange.ModifyWatermark;
import org.apache.flink.table.catalog.TableChange.ResetOption;
import org.apache.flink.table.catalog.TableChange.SetOption;
+import org.apache.flink.table.catalog.WatermarkSpec;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
@@ -99,6 +100,7 @@ import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;
+import static org.apache.flink.table.descriptors.DescriptorProperties.COMMENT;
import static org.apache.flink.table.descriptors.DescriptorProperties.NAME;
import static
org.apache.flink.table.descriptors.DescriptorProperties.WATERMARK;
import static
org.apache.flink.table.descriptors.DescriptorProperties.WATERMARK_ROWTIME;
@@ -119,8 +121,7 @@ import static
org.apache.paimon.flink.utils.FlinkCatalogPropertiesUtil.compoundK
import static
org.apache.paimon.flink.utils.FlinkCatalogPropertiesUtil.deserializeNonPhysicalColumn;
import static
org.apache.paimon.flink.utils.FlinkCatalogPropertiesUtil.deserializeWatermarkSpec;
import static
org.apache.paimon.flink.utils.FlinkCatalogPropertiesUtil.nonPhysicalColumnsCount;
-import static
org.apache.paimon.flink.utils.FlinkCatalogPropertiesUtil.serializeNonPhysicalColumns;
-import static
org.apache.paimon.flink.utils.FlinkCatalogPropertiesUtil.serializeWatermarkSpec;
+import static
org.apache.paimon.flink.utils.FlinkCatalogPropertiesUtil.serializeNewWatermarkSpec;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Catalog for paimon. */
@@ -670,6 +671,7 @@ public class FlinkCatalog extends AbstractCatalog {
Map<String, String> newOptions = new HashMap<>(table.options());
TableSchema.Builder builder = TableSchema.builder();
+ Map<String, String> nonPhysicalColumnComments = new HashMap<>();
// add columns
List<RowType.RowField> physicalRowFields =
toLogicalType(table.rowType()).getFields();
@@ -687,6 +689,11 @@ public class FlinkCatalog extends AbstractCatalog {
} else {
// build non-physical column from options
builder.add(deserializeNonPhysicalColumn(newOptions, i));
+ if (newOptions.containsKey(compoundKey(SCHEMA, i, COMMENT))) {
+ nonPhysicalColumnComments.put(
+ optionalName, newOptions.get(compoundKey(SCHEMA,
i, COMMENT)));
+ newOptions.remove(compoundKey(SCHEMA, i, COMMENT));
+ }
}
}
@@ -711,11 +718,17 @@ public class FlinkCatalog extends AbstractCatalog {
removeProperties.asMap().keySet().forEach(newOptions::remove);
return new DataCatalogTable(
- table, schema, table.partitionKeys(), newOptions,
table.comment().orElse(""));
+ table,
+ schema,
+ table.partitionKeys(),
+ newOptions,
+ table.comment().orElse(""),
+ nonPhysicalColumnComments);
}
- public static Schema fromCatalogTable(CatalogTable catalogTable) {
- TableSchema schema = catalogTable.getSchema();
+ public static Schema fromCatalogTable(CatalogTable table) {
+ ResolvedCatalogTable catalogTable = (ResolvedCatalogTable) table;
+ ResolvedSchema schema = catalogTable.getResolvedSchema();
RowType rowType = (RowType)
schema.toPhysicalRowDataType().getLogicalType();
Map<String, String> options = new HashMap<>(catalogTable.getOptions());
@@ -754,29 +767,14 @@ public class FlinkCatalog extends AbstractCatalog {
}
/** Only reserve necessary options. */
- private static Map<String, String> columnOptions(TableSchema schema) {
- Map<String, String> columnOptions = new HashMap<>();
- // field name -> index
- final Map<String, Integer> indexMap = new HashMap<>();
- List<TableColumn> tableColumns = schema.getTableColumns();
- for (int i = 0; i < tableColumns.size(); i++) {
- indexMap.put(tableColumns.get(i).getName(), i);
- }
-
- // non-physical columns
- List<TableColumn> nonPhysicalColumns =
- tableColumns.stream().filter(c ->
!c.isPhysical()).collect(Collectors.toList());
- if (!nonPhysicalColumns.isEmpty()) {
- columnOptions.putAll(serializeNonPhysicalColumns(indexMap,
nonPhysicalColumns));
- }
-
- // watermark
+ private static Map<String, String> columnOptions(ResolvedSchema schema) {
+ Map<String, String> columnOptions =
+ new
HashMap<>(FlinkCatalogPropertiesUtil.serializeNonPhysicalNewColumns(schema));
List<WatermarkSpec> watermarkSpecs = schema.getWatermarkSpecs();
if (!watermarkSpecs.isEmpty()) {
checkArgument(watermarkSpecs.size() == 1);
-
columnOptions.putAll(serializeWatermarkSpec(watermarkSpecs.get(0)));
+
columnOptions.putAll(serializeNewWatermarkSpec(watermarkSpecs.get(0)));
}
-
return columnOptions;
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/FlinkCatalogPropertiesUtil.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/FlinkCatalogPropertiesUtil.java
index fb019c6e4..b0f99a6e8 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/FlinkCatalogPropertiesUtil.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/FlinkCatalogPropertiesUtil.java
@@ -36,6 +36,7 @@ import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static org.apache.flink.table.descriptors.DescriptorProperties.COMMENT;
import static
org.apache.flink.table.descriptors.DescriptorProperties.DATA_TYPE;
import static org.apache.flink.table.descriptors.DescriptorProperties.EXPR;
import static org.apache.flink.table.descriptors.DescriptorProperties.METADATA;
@@ -97,6 +98,10 @@ public class FlinkCatalogPropertiesUtil {
serialized.put(
compoundKey(SCHEMA, index, EXPR),
computedColumn.getExpression().asSerializableString());
+ if (computedColumn.getComment().isPresent()) {
+ serialized.put(
+ compoundKey(SCHEMA, index, COMMENT),
computedColumn.getComment().get());
+ }
} else {
Column.MetadataColumn metadataColumn = (Column.MetadataColumn)
c;
serialized.put(
@@ -105,6 +110,10 @@ public class FlinkCatalogPropertiesUtil {
serialized.put(
compoundKey(SCHEMA, index, VIRTUAL),
Boolean.toString(metadataColumn.isVirtual()));
+ if (metadataColumn.getComment().isPresent()) {
+ serialized.put(
+ compoundKey(SCHEMA, index, COMMENT),
metadataColumn.getComment().get());
+ }
}
}
return serialized;
@@ -126,6 +135,27 @@ public class FlinkCatalogPropertiesUtil {
return serializedWatermarkSpec;
}
+ public static Map<String, String> serializeNewWatermarkSpec(
+ org.apache.flink.table.catalog.WatermarkSpec watermarkSpec) {
+ Map<String, String> serializedWatermarkSpec = new HashMap<>();
+ String watermarkPrefix = compoundKey(SCHEMA, WATERMARK, 0);
+ serializedWatermarkSpec.put(
+ compoundKey(watermarkPrefix, WATERMARK_ROWTIME),
+ watermarkSpec.getRowtimeAttribute());
+ serializedWatermarkSpec.put(
+ compoundKey(watermarkPrefix, WATERMARK_STRATEGY_EXPR),
+ watermarkSpec.getWatermarkExpression().asSerializableString());
+ serializedWatermarkSpec.put(
+ compoundKey(watermarkPrefix, WATERMARK_STRATEGY_DATA_TYPE),
+ watermarkSpec
+ .getWatermarkExpression()
+ .getOutputDataType()
+ .getLogicalType()
+ .asSerializableString());
+
+ return serializedWatermarkSpec;
+ }
+
private static final Pattern SCHEMA_COLUMN_NAME_SUFFIX =
Pattern.compile("\\d+\\." + NAME);
private static final Pattern SCHEMA_COLUMN_METADATA_SUFFIX =
Pattern.compile("\\d+\\." + METADATA);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
index 234656ee9..ab798c426 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
@@ -774,6 +774,20 @@ public class CatalogTableITCase extends CatalogITCaseBase {
.doesNotContain("schema");
}
+ @Test
+ public void testShowTableMetadataComment() {
+ sql("CREATE TABLE T (a INT, name VARCHAR METADATA COMMENT 'header1', b
INT)");
+ List<Row> result = sql("SHOW CREATE TABLE T");
+ assertThat(result.get(0).toString())
+ .contains(
+ "CREATE TABLE `PAIMON`.`default`.`T` (\n"
+ + " `a` INT,\n"
+ + " `name` VARCHAR(2147483647) METADATA
COMMENT 'header1',\n"
+ + " `b` INT\n"
+ + ")")
+ .doesNotContain("schema");
+ }
+
@Test
public void testReadOptimizedTable() {
sql("CREATE TABLE T (k INT, v INT, PRIMARY KEY (k) NOT ENFORCED)");
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
index 7f60a2b67..6fcb87f0c 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
@@ -29,20 +29,21 @@ import org.apache.paimon.flink.log.LogStoreTableFactory;
import org.apache.paimon.fs.Path;
import org.apache.paimon.options.Options;
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
+
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.TableDescriptor;
-import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogBaseTable;
import org.apache.flink.table.catalog.CatalogDatabase;
import org.apache.flink.table.catalog.CatalogDatabaseImpl;
import org.apache.flink.table.catalog.CatalogTable;
-import org.apache.flink.table.catalog.CatalogTableImpl;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.ResolvedCatalogTable;
import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
@@ -51,6 +52,8 @@ import
org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.expressions.utils.ResolvedExpressionMock;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.BeforeEach;
@@ -460,24 +463,39 @@ public class FlinkCatalogTest {
@Test
public void testCreateTableWithColumnOptions() throws Exception {
- TableSchema schema =
- TableSchema.builder()
- .field("pk", DataTypes.INT().notNull())
- .field("test", DataTypes.INT())
- .field("comp", DataTypes.INT(), "test + 1")
+ ResolvedExpression expression =
+ new ResolvedExpressionMock(DataTypes.INT(), () -> "test + 1");
+ ResolvedSchema resolvedSchema =
+ new ResolvedSchema(
+ Arrays.asList(
+ Column.physical("pk",
DataTypes.INT().notNull()),
+ Column.physical("test", DataTypes.INT()),
+ Column.computed("comp", expression)),
+ Collections.emptyList(),
+ UniqueConstraint.primaryKey("pk",
ImmutableList.of("pk")));
+
+ Schema schema =
+ Schema.newBuilder()
+ .column("pk", DataTypes.INT().notNull())
+ .column("test", DataTypes.INT())
+ .columnByExpression("comp", "test + 1")
.primaryKey("pk")
.build();
- CatalogTable catalogTable = new CatalogTableImpl(schema, new
HashMap<>(), "");
+
+ CatalogTable catalogTable =
+ new ResolvedCatalogTable(
+ CatalogTable.of(schema, "", Collections.emptyList(),
new HashMap<>()),
+ resolvedSchema);
catalog.createDatabase(path1.getDatabaseName(), null, false);
catalog.createTable(path1, catalogTable, false);
CatalogTable got = (CatalogTable) catalog.getTable(path1);
- TableSchema newSchema = got.getSchema();
+ Schema newSchema = got.getUnresolvedSchema();
-
assertThat(schema.getTableColumns()).isEqualTo(newSchema.getTableColumns());
- assertThat(schema.getPrimaryKey().get().getColumns())
- .isEqualTo(newSchema.getPrimaryKey().get().getColumns());
+ assertThat(schema.getColumns()).isEqualTo(newSchema.getColumns());
+ assertThat(schema.getPrimaryKey().get().getColumnNames())
+ .isEqualTo(newSchema.getPrimaryKey().get().getColumnNames());
Map<String, String> expected = got.getOptions();
expected.remove("path");
@@ -489,21 +507,39 @@ public class FlinkCatalogTest {
public void testCreateTableWithLogSystemRegister() throws Exception {
catalog.createDatabase(path1.getDatabaseName(), null, false);
- TableSchema schema =
- TableSchema.builder()
- .field("pk", DataTypes.INT().notNull())
- .field("test", DataTypes.INT())
- .field("comp", DataTypes.INT(), "test + 1")
+ ResolvedExpression expression =
+ new ResolvedExpressionMock(DataTypes.INT(), () -> "test + 1");
+ ResolvedSchema resolvedSchema =
+ new ResolvedSchema(
+ Arrays.asList(
+ Column.physical("pk",
DataTypes.INT().notNull()),
+ Column.physical("test", DataTypes.INT()),
+ Column.computed("comp", expression)),
+ Collections.emptyList(),
+ UniqueConstraint.primaryKey("pk",
ImmutableList.of("pk")));
+
+ Schema schema =
+ Schema.newBuilder()
+ .column("pk", DataTypes.INT().notNull())
+ .column("test", DataTypes.INT())
+ .columnByExpression("comp", "test + 1")
.primaryKey("pk")
.build();
+
Map<String, String> options = new HashMap<>();
- CatalogTable catalogTable1 = new CatalogTableImpl(schema, options, "");
+ CatalogTable catalogTable1 =
+ new ResolvedCatalogTable(
+ CatalogTable.of(schema, "", Collections.emptyList(),
options),
+ resolvedSchema);
catalog.createTable(path1, catalogTable1, false);
CatalogBaseTable storedTable1 = catalog.getTable(path1);
assertThat(storedTable1.getOptions().containsKey("testing.log.store.topic")).isFalse();
options.put(LOG_SYSTEM.key(), TESTING_LOG_STORE);
- CatalogTable catalogTable2 = new CatalogTableImpl(schema, options, "");
+ CatalogTable catalogTable2 =
+ new ResolvedCatalogTable(
+ CatalogTable.of(schema, "", Collections.emptyList(),
options),
+ resolvedSchema);
catalog.createTable(path3, catalogTable2, false);
CatalogBaseTable storedTable2 = catalog.getTable(path3);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
index 5a634fa2e..aa9e41867 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
@@ -1226,7 +1226,7 @@ public class ReadWriteTableITCase extends
AbstractTestBase {
}
@Test
- public void testUnsupportedComputedColumnComments() {
+ public void testComputedColumnComments() {
String ddl = "CREATE TABLE T(a INT , b INT, c AS a + b COMMENT
'computed');";
bEnv.executeSql(ddl);
@@ -1236,9 +1236,9 @@ public class ReadWriteTableITCase extends
AbstractTestBase {
.collect(Collectors.toList());
assertThat(result)
.containsExactlyInAnyOrder(
- "+I[a, INT, true, null, null, null]",
- "+I[b, INT, true, null, null, null]",
- "+I[c, INT, true, null, AS `a` + `b`, null]");
+ "+I[a, INT, true, null, null, null, null]",
+ "+I[b, INT, true, null, null, null, null]",
+ "+I[c, INT, true, null, AS `a` + `b`, null,
computed]");
}
@Test
