[#3980]feat(flink-connector): Support partition for hive table operation by coolderli · Pull Request #4096 · apache/gravitino

coolderli · 2024-07-07T13:42:51Z

add UTs

What changes were proposed in this pull request?

support partitions for creating the hive table

Why are the changes needed?

Fix: [Subtask][flink-connector] Support partition for hive table operation #3980

Does this PR introduce any user-facing change?

no

How was this patch tested?

add ITs

coolderli · 2024-07-07T13:49:43Z

@FANNG1 Please help review this. Thanks.

flink-connector/src/main/java/com/datastrato/gravitino/flink/connector/catalog/BaseCatalog.java

flink-connector/src/main/java/com/datastrato/gravitino/flink/connector/TransformConverter.java

.../java/com/datastrato/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT.java

flink-connector/src/main/java/com/datastrato/gravitino/flink/connector/catalog/BaseCatalog.java

...nector/src/main/java/com/datastrato/gravitino/flink/connector/DefaultTransformConverter.java

FANNG1 · 2024-07-11T08:45:00Z

LGTM, except few comments

coolderli · 2024-07-11T12:38:49Z

@FANNG1 I fixed it. Please take a look.

FANNG1 · 2024-07-15T00:40:01Z

@coolderli , seems you add some partittion interfaces support since last review like listPartitions, is it expected?

flink-connector/src/main/java/com/datastrato/gravitino/flink/connector/catalog/BaseCatalog.java

FANNG1 · 2024-07-17T14:09:23Z

@coolderli , could you rebase the code?

…operation add UTs

coolderli · 2024-07-19T02:18:41Z

@coolderli , seems you add some partittion interfaces support since last review like listPartitions, is it expected?

@FANNG1 I think it's necessary. I meet an exception as follows. So I have to implement it.

FlinkHiveCatalogIT > testHivePartitionTable() FAILED
    java.lang.UnsupportedOperationException
        at org.apache.gravitino.flink.connector.catalog.BaseCatalog.listPartitions(BaseCatalog.java:292)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkRecomputeStatisticsProgram.getPartitionsTableStats(FlinkRecomputeStatisticsProgram.java:180)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkRecomputeStatisticsProgram.recomputeStatistics(FlinkRecomputeStatisticsProgram.java:144)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkRecomputeStatisticsProgram.recomputeStatistics(FlinkRecomputeStatisticsProgram.java:100)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkRecomputeStatisticsProgram.access$000(FlinkRecomputeStatisticsProgram.java:66)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkRecomputeStatisticsProgram$1.visit(FlinkRecomputeStatisticsProgram.java:75)
        at org.apache.calcite.rel.core.TableScan.accept(TableScan.java:180)
        at org.apache.flink.table.planner.plan.utils.DefaultRelShuttle.$anonfun$visit$1(RelShuttles.scala:37)
        at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233)
        at scala.collection.Iterator.foreach(Iterator.scala:937)
        at scala.collection.Iterator.foreach$(Iterator.scala:937)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
        at scala.collection.IterableLike.foreach(IterableLike.scala:70)
        at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
        at scala.collection.TraversableLike.map(TraversableLike.scala:233)
        at scala.collection.TraversableLike.map$(TraversableLike.scala:226)
        at scala.collection.AbstractTraversable.map(Traversable.scala:104)
        at org.apache.flink.table.planner.plan.utils.DefaultRelShuttle.visit(RelShuttles.scala:35)
        at org.apache.calcite.rel.RelHomogeneousShuttle.visit(RelHomogeneousShuttle.java:66)
        at org.apache.calcite.rel.logical.LogicalProject.accept(LogicalProject.java:135)
        at org.apache.flink.table.planner.plan.utils.DefaultRelShuttle.$anonfun$visit$1(RelShuttles.scala:37)
        at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233)
        at scala.collection.Iterator.foreach(Iterator.scala:937)
        at scala.collection.Iterator.foreach$(Iterator.scala:937)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
        at scala.collection.IterableLike.foreach(IterableLike.scala:70)
        at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
        at scala.collection.TraversableLike.map(TraversableLike.scala:233)
        at scala.collection.TraversableLike.map$(TraversableLike.scala:226)
        at scala.collection.AbstractTraversable.map(Traversable.scala:104)
        at org.apache.flink.table.planner.plan.utils.DefaultRelShuttle.visit(RelShuttles.scala:35)
        at org.apache.calcite.rel.RelHomogeneousShuttle.visit(RelHomogeneousShuttle.java:90)
        at org.apache.calcite.rel.logical.LogicalSort.accept(LogicalSort.java:86)
        at org.apache.flink.table.planner.plan.utils.DefaultRelShuttle.$anonfun$visit$1(RelShuttles.scala:37)
        at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233)
        at scala.collection.Iterator.foreach(Iterator.scala:937)
        at scala.collection.Iterator.foreach$(Iterator.scala:937)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
        at scala.collection.IterableLike.foreach(IterableLike.scala:70)
        at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
        at scala.collection.TraversableLike.map(TraversableLike.scala:233)
        at scala.collection.TraversableLike.map$(TraversableLike.scala:226)
        at scala.collection.AbstractTraversable.map(Traversable.scala:104)
        at org.apache.flink.table.planner.plan.utils.DefaultRelShuttle.visit(RelShuttles.scala:35)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkRecomputeStatisticsProgram.optimize(FlinkRecomputeStatisticsProgram.java:80)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkRecomputeStatisticsProgram.optimize(FlinkRecomputeStatisticsProgram.java:66)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:59)
        at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
        at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
        at scala.collection.Iterator.foreach(Iterator.scala:937)
        at scala.collection.Iterator.foreach$(Iterator.scala:937)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
        at scala.collection.IterableLike.foreach(IterableLike.scala:70)
        at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
        at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
        at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
        at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:56)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:51)
        at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
        at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
        at scala.collection.immutable.Range.foreach(Range.scala:155)
        at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
        at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
        at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:51)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:59)
        at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
        at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
        at scala.collection.Iterator.foreach(Iterator.scala:937)
        at scala.collection.Iterator.foreach$(Iterator.scala:937)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
        at scala.collection.IterableLike.foreach(IterableLike.scala:70)
        at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
        at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
        at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
        at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:55)
        at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.optimizeTree(BatchCommonSubGraphBasedOptimizer.scala:93)
        at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.optimizeBlock(BatchCommonSubGraphBasedOptimizer.scala:58)
        at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.$anonfun$doOptimize$1(BatchCommonSubGraphBasedOptimizer.scala:45)
        at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.$anonfun$doOptimize$1$adapted(BatchCommonSubGraphBasedOptimizer.scala:45)
        at scala.collection.immutable.List.foreach(List.scala:388)
        at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.doOptimize(BatchCommonSubGraphBasedOptimizer.scala:45)
        at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:87)
        at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:324)
        at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:182)
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1277)
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeQueryOperation(TableEnvironmentImpl.java:1055)
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1120)
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:735)
        at org.apache.gravitino.flink.connector.integration.test.FlinkEnvIT.sql(FlinkEnvIT.java:159)
        at org.apache.gravitino.flink.connector.integration.test.hive.FlinkHiveCatalogIT.lambda$testHivePartitionTable$3(FlinkHiveCatalogIT.java:360)
        at org.apache.gravitino.flink.connector.integration.test.FlinkEnvIT.doWithSchema(FlinkEnvIT.java:175)
        at org.apache.gravitino.flink.connector.integration.test.hive.FlinkHiveCatalogIT.testHivePartitionTable(FlinkHiveCatalogIT.java:307)

FANNG1 · 2024-07-19T07:15:30Z

flink-connector/src/main/java/org/apache/gravitino/flink/connector/PartitionConverter.java

+ * properties to store the partition transform, so we can implement this interface to achieve more
+ * partition transform.
+ */
+public abstract class PartitionConverter {


@FANNG1 I think the abstract class is OK. Other classes won't implement it. What do you think?

interface is more proper here to me

flink-connector/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java

coolderli · 2024-07-22T06:41:53Z

@FANNG1 I removed the getFunction and kept getXXStatistic. Please take a look. Thanks.

coolderli · 2024-07-22T06:52:40Z

getFunction is also needed.

FlinkHiveCatalogIT > testHivePartitionTable() FAILED
    org.apache.flink.table.api.ValidationException: SQL validation failed. null
        at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:200)
        at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:117)
        at org.apache.flink.table.planner.operations.SqlNodeToOperationConversion.convert(SqlNodeToOperationConversion.java:261)
        at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:106)
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:728)
        at org.apache.gravitino.flink.connector.integration.test.FlinkEnvIT.sql(FlinkEnvIT.java:159)
        at org.apache.gravitino.flink.connector.integration.test.hive.FlinkHiveCatalogIT.lambda$testHivePartitionTable$3(FlinkHiveCatalogIT.java:360)
        at org.apache.gravitino.flink.connector.integration.test.FlinkEnvIT.doWithSchema(FlinkEnvIT.java:175)
        at org.apache.gravitino.flink.connector.integration.test.hive.FlinkHiveCatalogIT.testHivePartitionTable(FlinkHiveCatalogIT.java:307)

        Caused by:
        java.lang.UnsupportedOperationException
            at org.apache.gravitino.flink.connector.catalog.BaseCatalog.getFunction(BaseCatalog.java:358)
            at org.apache.flink.table.catalog.FunctionCatalog.resolvePreciseFunctionReference(FunctionCatalog.java:617)
            at org.apache.flink.table.catalog.FunctionCatalog.resolvePreciseFunctionReference(FunctionCatalog.java:593)
            at org.apache.flink.table.catalog.FunctionCatalog.lambda$resolveAmbiguousFunctionReference$5(FunctionCatalog.java:671)
            at java.util.Optional.orElseGet(Optional.java:267)
            at org.apache.flink.table.catalog.FunctionCatalog.resolveAmbiguousFunctionReference(FunctionCatalog.java:671)
            at org.apache.flink.table.catalog.FunctionCatalog.lookupFunction(FunctionCatalog.java:424)
            at org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.lookupOperatorOverloads(FunctionCatalogOperatorTable.java:99)
            at org.apache.calcite.sql.util.ChainedSqlOperatorTable.lookupOperatorOverloads(ChainedSqlOperatorTable.java:69)
            at org.apache.calcite.sql.validate.SqlValidatorImpl.makeNullaryCall(SqlValidatorImpl.java:1829)
            at org.apache.calcite.sql.validate.SqlValidatorImpl$Expander.visit(SqlValidatorImpl.java:6430)
            at org.apache.calcite.sql.validate.SqlValidatorImpl$Expander.visit(SqlValidatorImpl.java:6414)
            at org.apache.calcite.sql.SqlIdentifier.accept(SqlIdentifier.java:324)
            at org.apache.calcite.sql.validate.SqlValidatorImpl$Expander.go(SqlValidatorImpl.java:6423)
            at org.apache.calcite.sql.validate.SqlValidatorImpl.expand(SqlValidatorImpl.java:6004)
            at org.apache.calcite.sql.validate.SqlValidatorImpl.validateOrderList(SqlValidatorImpl.java:4150)
            at org.apache.calcite.sql.validate.SqlValidatorImpl.validateSelect(SqlValidatorImpl.java:3621)
            at org.apache.calcite.sql.validate.SelectNamespace.validateImpl(SelectNamespace.java:64)
            at org.apache.calcite.sql.validate.AbstractNamespace.validate(AbstractNamespace.java:89)
            at org.apache.calcite.sql.validate.SqlValidatorImpl.validateNamespace(SqlValidatorImpl.java:1050)
            at org.apache.calcite.sql.validate.SqlValidatorImpl.validateQuery(SqlValidatorImpl.java:1025)
            at org.apache.calcite.sql.SqlSelect.validate(SqlSelect.java:248)
            at org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:1000)
            at org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:749)
            at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:196)

FANNG1 · 2024-07-22T12:14:27Z

@coolderli , merged to main, thanks for your contribution!

FANNG1 reviewed Jul 8, 2024

View reviewed changes

FANNG1 reviewed Jul 11, 2024

View reviewed changes

flink-connector/src/main/java/com/datastrato/gravitino/flink/connector/catalog/BaseCatalog.java Outdated Show resolved Hide resolved

...nector/src/main/java/com/datastrato/gravitino/flink/connector/DefaultTransformConverter.java Outdated Show resolved Hide resolved

FANNG1 reviewed Jul 15, 2024

View reviewed changes

flink-connector/src/main/java/com/datastrato/gravitino/flink/connector/catalog/BaseCatalog.java Show resolved Hide resolved

coolderli added 8 commits July 18, 2024 23:21

[apache#3980]feat(flink-connector): Support partition for hive table …

5e80748

…operation add UTs

remove useless

92247d9

refactor

61f9fef

fix

0f5a531

fix

141795a

fix

000d090

add uts

1841750

remove useless

c0849b0

coolderli force-pushed the issue-3980 branch from 4ac5332 to c0849b0 Compare July 18, 2024 15:28

fix

5ac3574

fix uts

06d3387

FANNG1 reviewed Jul 19, 2024

View reviewed changes

flink-connector/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java Show resolved Hide resolved

coolderli added 2 commits July 22, 2024 14:38

fix its

16bca30

remove useless

6ff2513

fix its

6f41320

FANNG1 approved these changes Jul 22, 2024

View reviewed changes

FANNG1 merged commit 529429e into apache:main Jul 22, 2024

Conversation

coolderli commented Jul 7, 2024

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

coolderli commented Jul 7, 2024

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

FANNG1 commented Jul 11, 2024

Uh oh!

coolderli commented Jul 11, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

FANNG1 commented Jul 15, 2024

Uh oh!

Uh oh!

FANNG1 commented Jul 17, 2024

Uh oh!

coolderli commented Jul 19, 2024

Uh oh!

FANNG1 Jul 19, 2024

Choose a reason for hiding this comment

Uh oh!

coolderli Jul 19, 2024

Choose a reason for hiding this comment

Uh oh!

FANNG1 Jul 20, 2024

Choose a reason for hiding this comment

Uh oh!

coolderli Jul 22, 2024

Choose a reason for hiding this comment

Uh oh!

Uh oh!

coolderli commented Jul 22, 2024

Uh oh!

coolderli commented Jul 22, 2024

Uh oh!

FANNG1 commented Jul 22, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

coolderli commented Jul 11, 2024 •

edited

Loading