[
https://issues.apache.org/jira/browse/BEAM-9379?focusedWorklogId=579436&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-579436
]
ASF GitHub Bot logged work on BEAM-9379:
----------------------------------------
Author: ASF GitHub Bot
Created on: 08/Apr/21 18:16
Start Date: 08/Apr/21 18:16
Worklog Time Spent: 10m
Work Description: ibzib commented on a change in pull request #13930:
URL: https://github.com/apache/beam/pull/13930#discussion_r609954370
##########
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -468,66 +440,174 @@ private static Expression value(
final Expression expression = list.append(list.newName("current"),
input);
- FieldType fromType = schema.getField(index).getType();
- Class convertTo = null;
- if (storageType == Object.class) {
- convertTo = Object.class;
- } else if (fromType.getTypeName().isLogicalType()) {
- convertTo =
LOGICAL_TYPE_TO_BASE_TYPE_MAP.get(fromType.getLogicalType().getIdentifier());
- } else {
- convertTo = TYPE_CONVERSION_MAP.get(fromType.getTypeName());
- }
- if (convertTo == null) {
- throw new UnsupportedOperationException("Unable to get " +
fromType.getTypeName());
+ FieldType fieldType = schema.getField(index).getType();
+ Expression value;
+ switch (fieldType.getTypeName()) {
+ case BYTE:
+ value = Expressions.call(expression, "getByte",
Expressions.constant(index));
Review comment:
Can you add a comment explaining that these are method calls on the
input Row? (It took me a minute to figure that out.)
##########
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -375,7 +374,7 @@ private static Expression castOutputTime(Expression value,
FieldType toType) {
// Convert TIME to LocalTime
if (value.getType() == java.sql.Time.class) {
valueDateTime = Expressions.call(BuiltInMethod.TIME_TO_INT.method,
valueDateTime);
- } else if (value.getType() == Long.class) {
+ } else if (value.getType() == Integer.class || value.getType() ==
Long.class) {
Review comment:
Why does this need to include Integer now?
##########
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -468,66 +440,174 @@ private static Expression value(
final Expression expression = list.append(list.newName("current"),
input);
- FieldType fromType = schema.getField(index).getType();
- Class convertTo = null;
- if (storageType == Object.class) {
- convertTo = Object.class;
- } else if (fromType.getTypeName().isLogicalType()) {
- convertTo =
LOGICAL_TYPE_TO_BASE_TYPE_MAP.get(fromType.getLogicalType().getIdentifier());
- } else {
- convertTo = TYPE_CONVERSION_MAP.get(fromType.getTypeName());
- }
- if (convertTo == null) {
- throw new UnsupportedOperationException("Unable to get " +
fromType.getTypeName());
+ FieldType fieldType = schema.getField(index).getType();
+ Expression value;
+ switch (fieldType.getTypeName()) {
+ case BYTE:
+ value = Expressions.call(expression, "getByte",
Expressions.constant(index));
+ break;
+ case INT16:
+ value = Expressions.call(expression, "getInt16",
Expressions.constant(index));
+ break;
+ case INT32:
+ value = Expressions.call(expression, "getInt32",
Expressions.constant(index));
+ break;
+ case INT64:
+ value = Expressions.call(expression, "getInt64",
Expressions.constant(index));
+ break;
+ case DECIMAL:
+ value = Expressions.call(expression, "getDecimal",
Expressions.constant(index));
+ break;
+ case FLOAT:
+ value = Expressions.call(expression, "getFloat",
Expressions.constant(index));
+ break;
+ case DOUBLE:
+ value = Expressions.call(expression, "getDouble",
Expressions.constant(index));
+ break;
+ case STRING:
+ value = Expressions.call(expression, "getString",
Expressions.constant(index));
+ break;
+ case DATETIME:
+ value = Expressions.call(expression, "getDateTime",
Expressions.constant(index));
+ break;
+ case BOOLEAN:
+ value = Expressions.call(expression, "getBoolean",
Expressions.constant(index));
+ break;
+ case BYTES:
+ value = Expressions.call(expression, "getBytes",
Expressions.constant(index));
+ break;
+ case ARRAY:
+ value = Expressions.call(expression, "getArray",
Expressions.constant(index));
+ if (storageType == Object.class
+ &&
TypeName.ROW.equals(fieldType.getCollectionElementType().getTypeName())) {
+ // Workaround for missing row output support
+ return Expressions.convert_(value, Object.class);
+ }
+ break;
+ case MAP:
+ value = Expressions.call(expression, "getMap",
Expressions.constant(index));
+ break;
+ case ROW:
+ value = Expressions.call(expression, "getRow",
Expressions.constant(index));
+ break;
+ case LOGICAL_TYPE:
+ String identifier = fieldType.getLogicalType().getIdentifier();
+ if (CharType.IDENTIFIER.equals(identifier)) {
+ value = Expressions.call(expression, "getString",
Expressions.constant(index));
+ } else if (TimeWithLocalTzType.IDENTIFIER.equals(identifier)) {
+ value = Expressions.call(expression, "getDateTime",
Expressions.constant(index));
+ } else if (SqlTypes.DATE.getIdentifier().equals(identifier)) {
+ value =
+ Expressions.convert_(
+ Expressions.call(
+ expression,
+ "getLogicalTypeValue",
+ Expressions.constant(index),
+ Expressions.constant(LocalDate.class)),
+ LocalDate.class);
+ } else if (SqlTypes.TIME.getIdentifier().equals(identifier)) {
+ value =
+ Expressions.convert_(
+ Expressions.call(
+ expression,
+ "getLogicalTypeValue",
+ Expressions.constant(index),
+ Expressions.constant(LocalTime.class)),
+ LocalTime.class);
+ } else if (SqlTypes.DATETIME.getIdentifier().equals(identifier)) {
+ value =
+ Expressions.convert_(
+ Expressions.call(
+ expression,
+ "getLogicalTypeValue",
+ Expressions.constant(index),
+ Expressions.constant(LocalDateTime.class)),
+ LocalDateTime.class);
+ } else {
+ throw new UnsupportedOperationException("Unable to get logical
type " + identifier);
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException("Unable to get " +
fieldType.getTypeName());
}
- Expression value =
- Expressions.convert_(
- Expressions.call(
- expression,
- "getBaseValue",
- Expressions.constant(index),
- Expressions.constant(convertTo)),
- convertTo);
- return (storageType != Object.class) ? value(value, fromType) : value;
+ return value(value, fieldType);
Review comment:
Nit: can we rename the `value` method(s) while we're here? It's
confusing to have a noun and a verb named the same thing here.
##########
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -468,66 +440,174 @@ private static Expression value(
final Expression expression = list.append(list.newName("current"),
input);
- FieldType fromType = schema.getField(index).getType();
- Class convertTo = null;
- if (storageType == Object.class) {
- convertTo = Object.class;
- } else if (fromType.getTypeName().isLogicalType()) {
- convertTo =
LOGICAL_TYPE_TO_BASE_TYPE_MAP.get(fromType.getLogicalType().getIdentifier());
- } else {
- convertTo = TYPE_CONVERSION_MAP.get(fromType.getTypeName());
- }
- if (convertTo == null) {
- throw new UnsupportedOperationException("Unable to get " +
fromType.getTypeName());
+ FieldType fieldType = schema.getField(index).getType();
+ Expression value;
+ switch (fieldType.getTypeName()) {
+ case BYTE:
+ value = Expressions.call(expression, "getByte",
Expressions.constant(index));
+ break;
+ case INT16:
+ value = Expressions.call(expression, "getInt16",
Expressions.constant(index));
+ break;
+ case INT32:
+ value = Expressions.call(expression, "getInt32",
Expressions.constant(index));
+ break;
+ case INT64:
+ value = Expressions.call(expression, "getInt64",
Expressions.constant(index));
+ break;
+ case DECIMAL:
+ value = Expressions.call(expression, "getDecimal",
Expressions.constant(index));
+ break;
+ case FLOAT:
+ value = Expressions.call(expression, "getFloat",
Expressions.constant(index));
+ break;
+ case DOUBLE:
+ value = Expressions.call(expression, "getDouble",
Expressions.constant(index));
+ break;
+ case STRING:
+ value = Expressions.call(expression, "getString",
Expressions.constant(index));
+ break;
+ case DATETIME:
+ value = Expressions.call(expression, "getDateTime",
Expressions.constant(index));
+ break;
+ case BOOLEAN:
+ value = Expressions.call(expression, "getBoolean",
Expressions.constant(index));
+ break;
+ case BYTES:
+ value = Expressions.call(expression, "getBytes",
Expressions.constant(index));
+ break;
+ case ARRAY:
+ value = Expressions.call(expression, "getArray",
Expressions.constant(index));
+ if (storageType == Object.class
+ &&
TypeName.ROW.equals(fieldType.getCollectionElementType().getTypeName())) {
+ // Workaround for missing row output support
+ return Expressions.convert_(value, Object.class);
+ }
+ break;
+ case MAP:
+ value = Expressions.call(expression, "getMap",
Expressions.constant(index));
+ break;
+ case ROW:
+ value = Expressions.call(expression, "getRow",
Expressions.constant(index));
+ break;
+ case LOGICAL_TYPE:
+ String identifier = fieldType.getLogicalType().getIdentifier();
+ if (CharType.IDENTIFIER.equals(identifier)) {
+ value = Expressions.call(expression, "getString",
Expressions.constant(index));
+ } else if (TimeWithLocalTzType.IDENTIFIER.equals(identifier)) {
+ value = Expressions.call(expression, "getDateTime",
Expressions.constant(index));
+ } else if (SqlTypes.DATE.getIdentifier().equals(identifier)) {
+ value =
+ Expressions.convert_(
+ Expressions.call(
+ expression,
+ "getLogicalTypeValue",
+ Expressions.constant(index),
+ Expressions.constant(LocalDate.class)),
+ LocalDate.class);
+ } else if (SqlTypes.TIME.getIdentifier().equals(identifier)) {
+ value =
+ Expressions.convert_(
+ Expressions.call(
+ expression,
+ "getLogicalTypeValue",
+ Expressions.constant(index),
+ Expressions.constant(LocalTime.class)),
+ LocalTime.class);
+ } else if (SqlTypes.DATETIME.getIdentifier().equals(identifier)) {
+ value =
+ Expressions.convert_(
+ Expressions.call(
+ expression,
+ "getLogicalTypeValue",
+ Expressions.constant(index),
+ Expressions.constant(LocalDateTime.class)),
+ LocalDateTime.class);
+ } else {
+ throw new UnsupportedOperationException("Unable to get logical
type " + identifier);
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException("Unable to get " +
fieldType.getTypeName());
}
- Expression value =
- Expressions.convert_(
- Expressions.call(
- expression,
- "getBaseValue",
- Expressions.constant(index),
- Expressions.constant(convertTo)),
- convertTo);
- return (storageType != Object.class) ? value(value, fromType) : value;
+ return value(value, fieldType);
}
- private static Expression value(Expression value, Schema.FieldType type) {
- if (type.getTypeName().isLogicalType()) {
- String logicalId = type.getLogicalType().getIdentifier();
- if (SqlTypes.TIME.getIdentifier().equals(logicalId)) {
+ private static Expression value(Expression value, FieldType fieldType) {
Review comment:
Can we rename this method to differentiate it from `value(BlockBuilder
list, int index, Type storageType, Expression input, Schema schema)` and better
describe what it's actually doing?
##########
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -468,66 +440,174 @@ private static Expression value(
final Expression expression = list.append(list.newName("current"),
input);
- FieldType fromType = schema.getField(index).getType();
- Class convertTo = null;
- if (storageType == Object.class) {
- convertTo = Object.class;
- } else if (fromType.getTypeName().isLogicalType()) {
- convertTo =
LOGICAL_TYPE_TO_BASE_TYPE_MAP.get(fromType.getLogicalType().getIdentifier());
- } else {
- convertTo = TYPE_CONVERSION_MAP.get(fromType.getTypeName());
- }
- if (convertTo == null) {
- throw new UnsupportedOperationException("Unable to get " +
fromType.getTypeName());
+ FieldType fieldType = schema.getField(index).getType();
+ Expression value;
+ switch (fieldType.getTypeName()) {
+ case BYTE:
+ value = Expressions.call(expression, "getByte",
Expressions.constant(index));
+ break;
+ case INT16:
+ value = Expressions.call(expression, "getInt16",
Expressions.constant(index));
+ break;
+ case INT32:
+ value = Expressions.call(expression, "getInt32",
Expressions.constant(index));
+ break;
+ case INT64:
+ value = Expressions.call(expression, "getInt64",
Expressions.constant(index));
+ break;
+ case DECIMAL:
+ value = Expressions.call(expression, "getDecimal",
Expressions.constant(index));
+ break;
+ case FLOAT:
+ value = Expressions.call(expression, "getFloat",
Expressions.constant(index));
+ break;
+ case DOUBLE:
+ value = Expressions.call(expression, "getDouble",
Expressions.constant(index));
+ break;
+ case STRING:
+ value = Expressions.call(expression, "getString",
Expressions.constant(index));
+ break;
+ case DATETIME:
+ value = Expressions.call(expression, "getDateTime",
Expressions.constant(index));
+ break;
+ case BOOLEAN:
+ value = Expressions.call(expression, "getBoolean",
Expressions.constant(index));
+ break;
+ case BYTES:
+ value = Expressions.call(expression, "getBytes",
Expressions.constant(index));
+ break;
+ case ARRAY:
+ value = Expressions.call(expression, "getArray",
Expressions.constant(index));
+ if (storageType == Object.class
+ &&
TypeName.ROW.equals(fieldType.getCollectionElementType().getTypeName())) {
+ // Workaround for missing row output support
+ return Expressions.convert_(value, Object.class);
+ }
+ break;
+ case MAP:
+ value = Expressions.call(expression, "getMap",
Expressions.constant(index));
+ break;
+ case ROW:
+ value = Expressions.call(expression, "getRow",
Expressions.constant(index));
+ break;
+ case LOGICAL_TYPE:
+ String identifier = fieldType.getLogicalType().getIdentifier();
+ if (CharType.IDENTIFIER.equals(identifier)) {
+ value = Expressions.call(expression, "getString",
Expressions.constant(index));
+ } else if (TimeWithLocalTzType.IDENTIFIER.equals(identifier)) {
+ value = Expressions.call(expression, "getDateTime",
Expressions.constant(index));
+ } else if (SqlTypes.DATE.getIdentifier().equals(identifier)) {
+ value =
+ Expressions.convert_(
+ Expressions.call(
+ expression,
+ "getLogicalTypeValue",
+ Expressions.constant(index),
+ Expressions.constant(LocalDate.class)),
+ LocalDate.class);
+ } else if (SqlTypes.TIME.getIdentifier().equals(identifier)) {
+ value =
+ Expressions.convert_(
+ Expressions.call(
+ expression,
+ "getLogicalTypeValue",
+ Expressions.constant(index),
+ Expressions.constant(LocalTime.class)),
+ LocalTime.class);
+ } else if (SqlTypes.DATETIME.getIdentifier().equals(identifier)) {
+ value =
+ Expressions.convert_(
+ Expressions.call(
+ expression,
+ "getLogicalTypeValue",
+ Expressions.constant(index),
+ Expressions.constant(LocalDateTime.class)),
+ LocalDateTime.class);
+ } else {
+ throw new UnsupportedOperationException("Unable to get logical
type " + identifier);
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException("Unable to get " +
fieldType.getTypeName());
}
- Expression value =
- Expressions.convert_(
- Expressions.call(
- expression,
- "getBaseValue",
- Expressions.constant(index),
- Expressions.constant(convertTo)),
- convertTo);
- return (storageType != Object.class) ? value(value, fromType) : value;
+ return value(value, fieldType);
}
- private static Expression value(Expression value, Schema.FieldType type) {
- if (type.getTypeName().isLogicalType()) {
- String logicalId = type.getLogicalType().getIdentifier();
- if (SqlTypes.TIME.getIdentifier().equals(logicalId)) {
+ private static Expression value(Expression value, FieldType fieldType) {
+ switch (fieldType.getTypeName()) {
+ case BYTE:
+ return Expressions.convert_(value, Byte.class);
Review comment:
Why are these conversions necessary? e.g. doesn't `getByte` already
return `Byte`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 579436)
Time Spent: 10h 50m (was: 10h 40m)
> Upgrade to Calcite 1.26.0
> -------------------------
>
> Key: BEAM-9379
> URL: https://issues.apache.org/jira/browse/BEAM-9379
> Project: Beam
> Issue Type: Task
> Components: dsl-sql
> Reporter: Rui Wang
> Assignee: Andrew Pilloud
> Priority: P2
> Time Spent: 10h 50m
> Remaining Estimate: 0h
>
> Upgrade to Calcite 1.22.0 after it gets released (expected by end of Feb
> 2020).
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
