Skip to content

Commit 4fc5018

Browse files
mergify[bot]miomiocat
mergify[bot]
and
miomiocat
authored
[Enhancement] Support complex type for paimon table (backport #66784) (backport #68998) (#69017)
Signed-off-by: miomiocat <284487410@qq.com> Co-authored-by: miomiocat <284487410@qq.com>
1 parent 6079ceb commit 4fc5018

4 files changed

Lines changed: 332 additions & 24 deletions

File tree

‎fe/fe-core/src/main/java/com/starrocks/catalog/PaimonTable.java‎

Lines changed: 7 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -14,15 +14,14 @@
1414

1515
package com.starrocks.catalog;
1616

17+
import com.google.common.base.Strings;
1718
import com.starrocks.analysis.DescriptorTable;
1819
import com.starrocks.common.util.TimeUtils;
20+
import com.starrocks.connector.paimon.PaimonUtils;
1921
import com.starrocks.planner.PaimonScanNode;
20-
import com.starrocks.thrift.TIcebergSchema;
21-
import com.starrocks.thrift.TIcebergSchemaField;
2222
import com.starrocks.thrift.TPaimonTable;
2323
import com.starrocks.thrift.TTableDescriptor;
2424
import com.starrocks.thrift.TTableType;
25-
import org.apache.paimon.catalog.Identifier;
2625
import org.apache.paimon.table.DataTable;
2726
import org.apache.paimon.types.DataField;
2827

@@ -40,6 +39,7 @@ public class PaimonTable extends Table {
4039
private String databaseName;
4140
private String tableName;
4241
private org.apache.paimon.table.Table paimonNativeTable;
42+
private String uuid;
4343
private List<String> partColumnNames;
4444
private List<String> paimonFieldNames;
4545
private Map<String, String> properties;
@@ -87,11 +87,10 @@ public void setPaimonNativeTable(org.apache.paimon.table.Table paimonNativeTable
8787

8888
@Override
8989
public String getUUID() {
90-
if (!new Identifier(databaseName, tableName).isSystemTable()) {
91-
return String.join(".", catalogName, paimonNativeTable.uuid());
92-
} else {
93-
return String.join(".", catalogName, databaseName, tableName, paimonNativeTable.uuid());
90+
if (Strings.isNullOrEmpty(this.uuid)) {
91+
this.uuid = String.join(".", catalogName, databaseName, tableName, paimonNativeTable.uuid().replace(".", "_"));
9492
}
93+
return this.uuid;
9594
}
9695

9796
@Override
@@ -147,29 +146,14 @@ public TTableDescriptor toThrift(List<DescriptorTable.ReferencedPartitionInfo> p
147146
tPaimonTable.setPaimon_native_table(encodedTable);
148147
tPaimonTable.setTime_zone(TimeUtils.getSessionTimeZone());
149148

150-
// reuse TIcebergSchema directly for compatibility.
151-
TIcebergSchema tPaimonSchema = new TIcebergSchema();
152-
List<DataField> paimonFields = paimonNativeTable.rowType().getFields();
153-
List<TIcebergSchemaField> tIcebergFields = new ArrayList<>(paimonFields.size());
154-
for (DataField field : paimonFields) {
155-
tIcebergFields.add(getTIcebergSchemaField(field));
156-
}
157-
tPaimonSchema.setFields(tIcebergFields);
158-
tPaimonTable.setPaimon_schema(tPaimonSchema);
149+
tPaimonTable.setPaimon_schema(PaimonUtils.getTPaimonSchema(this.paimonNativeTable.rowType()));
159150

160151
TTableDescriptor tTableDescriptor = new TTableDescriptor(id, TTableType.PAIMON_TABLE,
161152
fullSchema.size(), 0, tableName, databaseName);
162153
tTableDescriptor.setPaimonTable(tPaimonTable);
163154
return tTableDescriptor;
164155
}
165156

166-
private TIcebergSchemaField getTIcebergSchemaField(DataField field) {
167-
TIcebergSchemaField tPaimonSchemaField = new TIcebergSchemaField();
168-
tPaimonSchemaField.setField_id(field.id());
169-
tPaimonSchemaField.setName(field.name());
170-
return tPaimonSchemaField;
171-
}
172-
173157
@Override
174158
public boolean equals(Object o) {
175159
if (this == o) {

‎fe/fe-core/src/main/java/com/starrocks/connector/paimon/PaimonUtils.java‎

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,87 @@
1+
// Copyright 2021-present StarRocks, Inc. All rights reserved.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// https://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package com.starrocks.connector.paimon;
16+
17+
import com.starrocks.thrift.TIcebergSchema;
18+
import com.starrocks.thrift.TIcebergSchemaField;
19+
import org.apache.paimon.format.parquet.ParquetSchemaConverter;
20+
import org.apache.paimon.table.SpecialFields;
21+
import org.apache.paimon.types.ArrayType;
22+
import org.apache.paimon.types.DataField;
23+
import org.apache.paimon.types.DataType;
24+
import org.apache.paimon.types.DataTypeRoot;
25+
import org.apache.paimon.types.MapType;
26+
import org.apache.paimon.types.RowType;
27+
28+
import java.util.ArrayList;
29+
import java.util.List;
30+
31+
public class PaimonUtils {
32+
33+
public static TIcebergSchema getTPaimonSchema(RowType rowType) {
34+
// reuse TIcebergSchema directly for compatibility.
35+
TIcebergSchema tPaimonSchema = new TIcebergSchema();
36+
List<DataField> paimonFields = rowType.getFields();
37+
List<TIcebergSchemaField> tIcebergFields = new ArrayList<>(paimonFields.size());
38+
for (DataField field : paimonFields) {
39+
tIcebergFields.add(getTPaimonSchemaField(field.name(), field.type(), field.id(), 0, -1));
40+
}
41+
tPaimonSchema.setFields(tIcebergFields);
42+
return tPaimonSchema;
43+
}
44+
45+
public static TIcebergSchemaField getTPaimonSchemaField(String name, DataType type, int fieldId, int depth, int parentId) {
46+
TIcebergSchemaField tPaimonSchemaField = new TIcebergSchemaField();
47+
if (parentId != -1) {
48+
tPaimonSchemaField.setField_id(parentId);
49+
} else {
50+
tPaimonSchemaField.setField_id(fieldId);
51+
}
52+
tPaimonSchemaField.setName(name);
53+
if (type.getTypeRoot() == DataTypeRoot.MAP) {
54+
org.apache.paimon.types.MapType mapType = (MapType) type;
55+
DataType keyType = mapType.getKeyType();
56+
DataType valueType = mapType.getValueType();
57+
int mapKeyFieldId = SpecialFields.getMapKeyFieldId(fieldId, depth + 1);
58+
int mapValueFieldId = SpecialFields.getMapValueFieldId(fieldId, depth + 1);
59+
List<TIcebergSchemaField> children = new ArrayList<>(2);
60+
children.add(getTPaimonSchemaField(ParquetSchemaConverter.MAP_KEY_NAME, keyType, fieldId, depth + 1, mapKeyFieldId));
61+
children.add(getTPaimonSchemaField(ParquetSchemaConverter.MAP_VALUE_NAME, valueType, fieldId,
62+
depth + 1, mapValueFieldId));
63+
tPaimonSchemaField.setChildren(children);
64+
}
65+
if (type.getTypeRoot() == DataTypeRoot.ARRAY) {
66+
org.apache.paimon.types.ArrayType arrayType = (ArrayType) type;
67+
DataType elementType = arrayType.getElementType();
68+
int elementId = SpecialFields.getArrayElementFieldId(fieldId, depth + 1);
69+
List<TIcebergSchemaField> children = new ArrayList<>(1);
70+
children.add(getTPaimonSchemaField(ParquetSchemaConverter.LIST_ELEMENT_NAME, elementType, fieldId,
71+
depth + 1, elementId));
72+
tPaimonSchemaField.setChildren(children);
73+
}
74+
// the parent id of row type is always -1, refer to: org.apache.paimon.format.parquet.ParquetSchemaConverter
75+
if (type.getTypeRoot() == DataTypeRoot.ROW) {
76+
RowType rowType = (RowType) type;
77+
List<DataField> childrenFields = rowType.getFields();
78+
List<TIcebergSchemaField> children = new ArrayList<>(rowType.getFieldCount());
79+
for (DataField childrenField : childrenFields) {
80+
children.add(getTPaimonSchemaField(childrenField.name(), childrenField.type(), childrenField.id(),
81+
depth + 1, -1));
82+
}
83+
tPaimonSchemaField.setChildren(children);
84+
}
85+
return tPaimonSchemaField;
86+
}
87+
}

‎fe/fe-core/src/test/java/com/starrocks/connector/paimon/PaimonMetadataTest.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -190,6 +190,10 @@ public void testGetTable(@Mocked FileStoreTable paimonNativeTable) throws Catalo
190190
result = new ArrayList<>(Collections.singleton("col1"));
191191
paimonNativeTable.location().toString();
192192
result = "hdfs://127.0.0.1:10000/paimon";
193+
paimonNativeTable.primaryKeys();
194+
result = List.of("col2");
195+
paimonNativeTable.uuid();
196+
result = "fake_uuid";
193197
}
194198
};
195199
com.starrocks.catalog.Table table = metadata.getTable(connectContext, "db1", "tbl1");
@@ -204,7 +208,7 @@ public void testGetTable(@Mocked FileStoreTable paimonNativeTable) throws Catalo
204208
assertEquals(ScalarType.DOUBLE, paimonTable.getBaseSchema().get(1).getType());
205209
org.junit.jupiter.api.Assertions.assertTrue(paimonTable.getBaseSchema().get(1).isAllowNull());
206210
assertEquals("paimon_catalog", paimonTable.getCatalogName());
207-
assertEquals("paimon_catalog.null", paimonTable.getUUID());
211+
assertEquals("paimon_catalog.db1.tbl1.fake_uuid", paimonTable.getUUID());
208212
}
209213

210214
@Test

0 commit comments

Comments
 (0)