Skip to content

Commit

Permalink
[Fix-3799] Fix the issue of array out of bounds when fetching lineage…
Browse files Browse the repository at this point in the history 
… information (#3816)

Co-authored-by: luoshangjie <[email protected]>
Co-authored-by: GH Action - Upstream Sync <[email protected]>
Co-authored-by: 18216499322 <[email protected]>
  • Loading branch information
@18216499322 @actions-user
4 people authored Sep 18, 2024
1 parent 43b6486 commit 2387318
Show file tree
Hide file tree
Showing 22 changed files with 349 additions and 65 deletions.
13 changes: 13 additions & 0 deletions ...ient/dinky-client-1.14/src/main/java/org/apache/calcite/rel/metadata/RelColumnOrigin.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,8 @@ public class RelColumnOrigin {

private final boolean isDerived;

private boolean isComputedColumn;

/**
* Stores the expression for data conversion,
* which source table fields are transformed by which expression the target field
Expand All @@ -63,6 +65,13 @@ public RelColumnOrigin(RelOptTable originTable, int iOriginColumn, boolean isDer
this.transform = transform;
}

public RelColumnOrigin(RelOptTable originTable, int iOriginColumn, boolean isDerived, boolean isComputedColumn) {
this.originTable = originTable;
this.iOriginColumn = iOriginColumn;
this.isDerived = isDerived;
this.isComputedColumn = isComputedColumn;
}

// ~ Methods ----------------------------------------------------------------

/**
Expand Down Expand Up @@ -94,6 +103,10 @@ public boolean isDerived() {
return isDerived;
}

public boolean isComputedColumn() {
return isComputedColumn;
}

public String getTransform() {
return transform;
}
Expand Down
24 changes: 20 additions & 4 deletions ...t/dinky-client-1.14/src/main/java/org/apache/calcite/rel/metadata/RelMdColumnOrigins.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,8 @@
import org.apache.calcite.rex.RexVisitor;
import org.apache.calcite.rex.RexVisitorImpl;
import org.apache.calcite.util.BuiltInMethod;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.planner.plan.schema.TableSourceTable;

import java.util.ArrayList;
import java.util.Collections;
Expand Down Expand Up @@ -230,14 +232,28 @@ public Set<RelColumnOrigin> getColumnOrigins(Project rel, final RelMetadataQuery
// Direct reference: no derivation added.
RexInputRef inputRef = (RexInputRef) rexNode;
int index = inputRef.getIndex();
if (input instanceof TableScan) {
index = computeIndexWithOffset(rel.getProjects(), inputRef.getIndex(), iOutputColumn);
}
return mq.getColumnOrigins(input, index);
} else if (input instanceof TableScan
&& rexNode.getClass().equals(RexCall.class)
&& ((RexCall) rexNode).getOperands().isEmpty()) {
return mq.getColumnOrigins(input, iOutputColumn);
List<Column> columns = ((TableSourceTable) (input).getTable())
.catalogTable()
.getResolvedSchema()
.getColumns();
Set<RelColumnOrigin> set = new LinkedHashSet<>();
for (int index = 0; index < columns.size(); index++) {
Column column = columns.get(index);
if (column instanceof Column.ComputedColumn
&& rexNode.toString()
.equals(((Column.ComputedColumn) column)
.getExpression()
.toString())) {
set.add(new RelColumnOrigin(input.getTable(), index, false, true));
return set;
}
}
set.add(new RelColumnOrigin(input.getTable(), -1, false, false));
return set;
}
// Anything else is a derivation, possibly from multiple columns.
final Set<RelColumnOrigin> set = getMultipleColumns(rexNode, input, mq);
Expand Down
21 changes: 16 additions & 5 deletions dinky-client/dinky-client-1.14/src/main/java/org/dinky/utils/LineageContext.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -128,11 +128,22 @@ private List<LineageRel> buildFiledLineageResult(String sinkTable, RelNode optRe

// filed
int ordinal = relColumnOrigin.getOriginColumnOrdinal();
List<String> fieldNames = ((TableSourceTable) table)
.catalogTable()
.getResolvedSchema()
.getColumnNames();
String sourceColumn = fieldNames.get(ordinal);

if (ordinal == -1) {
continue;
}

String sourceColumn;
if (relColumnOrigin.isComputedColumn()) {
List<String> fieldNames = ((TableSourceTable) table)
.catalogTable()
.getResolvedSchema()
.getColumnNames();
sourceColumn = fieldNames.get(ordinal);
} else {
List<String> fieldNames = table.getRowType().getFieldNames();
sourceColumn = fieldNames.get(ordinal);
}

// add record
resultList.add(LineageRel.build(
Expand Down
13 changes: 13 additions & 0 deletions ...ient/dinky-client-1.15/src/main/java/org/apache/calcite/rel/metadata/RelColumnOrigin.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,8 @@ public class RelColumnOrigin {

private final boolean isDerived;

private boolean isComputedColumn;

/**
* Stores the expression for data conversion,
* which source table fields are transformed by which expression the target field
Expand All @@ -63,6 +65,13 @@ public RelColumnOrigin(RelOptTable originTable, int iOriginColumn, boolean isDer
this.transform = transform;
}

public RelColumnOrigin(RelOptTable originTable, int iOriginColumn, boolean isDerived, boolean isComputedColumn) {
this.originTable = originTable;
this.iOriginColumn = iOriginColumn;
this.isDerived = isDerived;
this.isComputedColumn = isComputedColumn;
}

// ~ Methods ----------------------------------------------------------------

/**
Expand Down Expand Up @@ -98,6 +107,10 @@ public String getTransform() {
return transform;
}

public boolean isComputedColumn() {
return isComputedColumn;
}

@Override
public boolean equals(Object obj) {
if (!(obj instanceof RelColumnOrigin)) {
Expand Down
24 changes: 20 additions & 4 deletions ...t/dinky-client-1.15/src/main/java/org/apache/calcite/rel/metadata/RelMdColumnOrigins.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,8 @@
import org.apache.calcite.rex.RexVisitor;
import org.apache.calcite.rex.RexVisitorImpl;
import org.apache.calcite.util.BuiltInMethod;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.planner.plan.schema.TableSourceTable;

import java.util.ArrayList;
import java.util.Collections;
Expand Down Expand Up @@ -230,14 +232,28 @@ public Set<RelColumnOrigin> getColumnOrigins(Project rel, final RelMetadataQuery
// Direct reference: no derivation added.
RexInputRef inputRef = (RexInputRef) rexNode;
int index = inputRef.getIndex();
if (input instanceof TableScan) {
index = computeIndexWithOffset(rel.getProjects(), inputRef.getIndex(), iOutputColumn);
}
return mq.getColumnOrigins(input, index);
} else if (input instanceof TableScan
&& rexNode.getClass().equals(RexCall.class)
&& ((RexCall) rexNode).getOperands().isEmpty()) {
return mq.getColumnOrigins(input, iOutputColumn);
List<Column> columns = ((TableSourceTable) (input).getTable())
.contextResolvedTable()
.getResolvedSchema()
.getColumns();
Set<RelColumnOrigin> set = new LinkedHashSet<>();
for (int index = 0; index < columns.size(); index++) {
Column column = columns.get(index);
if (column instanceof Column.ComputedColumn
&& rexNode.toString()
.equals(((Column.ComputedColumn) column)
.getExpression()
.toString())) {
set.add(new RelColumnOrigin(input.getTable(), index, false, true));
return set;
}
}
set.add(new RelColumnOrigin(input.getTable(), -1, false, false));
return set;
}
// Anything else is a derivation, possibly from multiple columns.
final Set<RelColumnOrigin> set = getMultipleColumns(rexNode, input, mq);
Expand Down
21 changes: 16 additions & 5 deletions dinky-client/dinky-client-1.15/src/main/java/org/dinky/utils/LineageContext.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -131,11 +131,22 @@ private List<LineageRel> buildFiledLineageResult(String sinkTable, RelNode optRe

// filed
int ordinal = relColumnOrigin.getOriginColumnOrdinal();
List<String> fieldNames = ((TableSourceTable) table)
.contextResolvedTable()
.getResolvedSchema()
.getColumnNames();
String sourceColumn = fieldNames.get(ordinal);

if (ordinal == -1) {
continue;
}

String sourceColumn;
if (relColumnOrigin.isComputedColumn()) {
List<String> fieldNames = ((TableSourceTable) table)
.contextResolvedTable()
.getResolvedSchema()
.getColumnNames();
sourceColumn = fieldNames.get(ordinal);
} else {
List<String> fieldNames = table.getRowType().getFieldNames();
sourceColumn = fieldNames.get(ordinal);
}

// add record
resultList.add(LineageRel.build(
Expand Down
13 changes: 13 additions & 0 deletions ...ient/dinky-client-1.16/src/main/java/org/apache/calcite/rel/metadata/RelColumnOrigin.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,8 @@ public class RelColumnOrigin {

private final boolean isDerived;

private boolean isComputedColumn;

/**
* Stores the expression for data conversion,
* which source table fields are transformed by which expression the target field
Expand All @@ -63,6 +65,13 @@ public RelColumnOrigin(RelOptTable originTable, int iOriginColumn, boolean isDer
this.transform = transform;
}

public RelColumnOrigin(RelOptTable originTable, int iOriginColumn, boolean isDerived, boolean isComputedColumn) {
this.originTable = originTable;
this.iOriginColumn = iOriginColumn;
this.isDerived = isDerived;
this.isComputedColumn = isComputedColumn;
}

// ~ Methods ----------------------------------------------------------------

/**
Expand Down Expand Up @@ -98,6 +107,10 @@ public String getTransform() {
return transform;
}

public boolean isComputedColumn() {
return isComputedColumn;
}

@Override
public boolean equals(Object obj) {
if (!(obj instanceof RelColumnOrigin)) {
Expand Down
27 changes: 23 additions & 4 deletions ...t/dinky-client-1.16/src/main/java/org/apache/calcite/rel/metadata/RelMdColumnOrigins.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,8 @@
import org.apache.calcite.rex.RexVisitor;
import org.apache.calcite.rex.RexVisitorImpl;
import org.apache.calcite.util.BuiltInMethod;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.planner.plan.schema.TableSourceTable;

import java.util.ArrayList;
import java.util.Collections;
Expand Down Expand Up @@ -230,14 +232,31 @@ public Set<RelColumnOrigin> getColumnOrigins(Project rel, final RelMetadataQuery
// Direct reference: no derivation added.
RexInputRef inputRef = (RexInputRef) rexNode;
int index = inputRef.getIndex();
if (input instanceof TableScan) {
index = computeIndexWithOffset(rel.getProjects(), inputRef.getIndex(), iOutputColumn);
}
// if (input instanceof TableScan) {
// index = computeIndexWithOffset(rel.getProjects(), inputRef.getIndex(), iOutputColumn);
// }
return mq.getColumnOrigins(input, index);
} else if (input instanceof TableScan
&& rexNode.getClass().equals(RexCall.class)
&& ((RexCall) rexNode).getOperands().isEmpty()) {
return mq.getColumnOrigins(input, iOutputColumn);
List<Column> columns = ((TableSourceTable) (input).getTable())
.contextResolvedTable()
.getResolvedSchema()
.getColumns();
Set<RelColumnOrigin> set = new LinkedHashSet<>();
for (int index = 0; index < columns.size(); index++) {
Column column = columns.get(index);
if (column instanceof Column.ComputedColumn
&& rexNode.toString()
.equals(((Column.ComputedColumn) column)
.getExpression()
.toString())) {
set.add(new RelColumnOrigin(input.getTable(), index, false, true));
return set;
}
}
set.add(new RelColumnOrigin(input.getTable(), -1, false, false));
return set;
}
// Anything else is a derivation, possibly from multiple columns.
final Set<RelColumnOrigin> set = getMultipleColumns(rexNode, input, mq);
Expand Down
21 changes: 16 additions & 5 deletions dinky-client/dinky-client-1.16/src/main/java/org/dinky/utils/LineageContext.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -133,11 +133,22 @@ private List<LineageRel> buildFiledLineageResult(String sinkTable, RelNode optRe

// filed
int ordinal = relColumnOrigin.getOriginColumnOrdinal();
List<String> fieldNames = ((TableSourceTable) table)
.contextResolvedTable()
.getResolvedSchema()
.getColumnNames();
String sourceColumn = fieldNames.get(ordinal);

if (ordinal == -1) {
continue;
}

String sourceColumn;
if (relColumnOrigin.isComputedColumn()) {
List<String> fieldNames = ((TableSourceTable) table)
.contextResolvedTable()
.getResolvedSchema()
.getColumnNames();
sourceColumn = fieldNames.get(ordinal);
} else {
List<String> fieldNames = table.getRowType().getFieldNames();
sourceColumn = fieldNames.get(ordinal);
}

// add record
resultList.add(LineageRel.build(
Expand Down
13 changes: 13 additions & 0 deletions ...ient/dinky-client-1.17/src/main/java/org/apache/calcite/rel/metadata/RelColumnOrigin.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,8 @@ public class RelColumnOrigin {

private final boolean isDerived;

private boolean isComputedColumn;

/**
* Stores the expression for data conversion,
* which source table fields are transformed by which expression the target field
Expand All @@ -63,6 +65,13 @@ public RelColumnOrigin(RelOptTable originTable, int iOriginColumn, boolean isDer
this.transform = transform;
}

public RelColumnOrigin(RelOptTable originTable, int iOriginColumn, boolean isDerived, boolean isComputedColumn) {
this.originTable = originTable;
this.iOriginColumn = iOriginColumn;
this.isDerived = isDerived;
this.isComputedColumn = isComputedColumn;
}

// ~ Methods ----------------------------------------------------------------

/**
Expand Down Expand Up @@ -98,6 +107,10 @@ public String getTransform() {
return transform;
}

public boolean isComputedColumn() {
return isComputedColumn;
}

@Override
public boolean equals(Object obj) {
if (!(obj instanceof RelColumnOrigin)) {
Expand Down
24 changes: 20 additions & 4 deletions ...t/dinky-client-1.17/src/main/java/org/apache/calcite/rel/metadata/RelMdColumnOrigins.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,8 @@
import org.apache.calcite.rex.RexVisitor;
import org.apache.calcite.rex.RexVisitorImpl;
import org.apache.calcite.util.BuiltInMethod;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.planner.plan.schema.TableSourceTable;

import java.util.ArrayList;
import java.util.Collections;
Expand Down Expand Up @@ -230,14 +232,28 @@ public Set<RelColumnOrigin> getColumnOrigins(Project rel, final RelMetadataQuery
// Direct reference: no derivation added.
RexInputRef inputRef = (RexInputRef) rexNode;
int index = inputRef.getIndex();
if (input instanceof TableScan) {
index = computeIndexWithOffset(rel.getProjects(), inputRef.getIndex(), iOutputColumn);
}
return mq.getColumnOrigins(input, index);
} else if (input instanceof TableScan
&& rexNode.getClass().equals(RexCall.class)
&& ((RexCall) rexNode).getOperands().isEmpty()) {
return mq.getColumnOrigins(input, iOutputColumn);
List<Column> columns = ((TableSourceTable) (input).getTable())
.contextResolvedTable()
.getResolvedSchema()
.getColumns();
Set<RelColumnOrigin> set = new LinkedHashSet<>();
for (int index = 0; index < columns.size(); index++) {
Column column = columns.get(index);
if (column instanceof Column.ComputedColumn
&& rexNode.toString()
.equals(((Column.ComputedColumn) column)
.getExpression()
.toString())) {
set.add(new RelColumnOrigin(input.getTable(), index, false, true));
return set;
}
}
set.add(new RelColumnOrigin(input.getTable(), -1, false, false));
return set;
}
// Anything else is a derivation, possibly from multiple columns.
final Set<RelColumnOrigin> set = getMultipleColumns(rexNode, input, mq);
Expand Down
Loading

0 comments on commit 2387318

Please sign in to comment.