| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.beam.sdk.extensions.sql.zetasql; |
| |
| import static org.apache.beam.sdk.extensions.sql.zetasql.DateTimeUtils.parseDate; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.DateTimeUtils.parseDateToValue; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.DateTimeUtils.parseTime; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.DateTimeUtils.parseTimeToValue; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.DateTimeUtils.parseTimestampWithTZToValue; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.DateTimeUtils.parseTimestampWithTimeZone; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.DateTimeUtils.parseTimestampWithUTCTimeZone; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.AGGREGATE_TABLE_ONE; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.AGGREGATE_TABLE_TWO; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.BASIC_TABLE_ONE; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.BASIC_TABLE_THREE; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.BASIC_TABLE_TWO; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TABLE_ALL_NULL; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TABLE_ALL_TYPES; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TABLE_ALL_TYPES_2; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TABLE_EMPTY; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TABLE_FOR_CASE_WHEN; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TABLE_WITH_ARRAY; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TABLE_WITH_ARRAY_FOR_UNNEST; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TABLE_WITH_MAP; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TABLE_WITH_STRUCT; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TABLE_WITH_STRUCT_TIMESTAMP_STRING; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TABLE_WITH_STRUCT_TWO; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TIMESTAMP_TABLE_ONE; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TIMESTAMP_TABLE_TWO; |
| import static org.apache.beam.sdk.extensions.sql.zetasql.TestInput.TIME_TABLE; |
| import static org.apache.beam.sdk.schemas.Schema.FieldType.DATETIME; |
| |
| import com.google.protobuf.ByteString; |
| import com.google.zetasql.SqlException; |
| import com.google.zetasql.StructType.StructField; |
| import com.google.zetasql.TypeFactory; |
| import com.google.zetasql.Value; |
| import com.google.zetasql.ZetaSQLType.TypeKind; |
| import com.google.zetasql.ZetaSQLValue.ValueProto; |
| import java.nio.charset.StandardCharsets; |
| import java.util.Arrays; |
| import java.util.HashMap; |
| import java.util.Map; |
| import org.apache.beam.sdk.extensions.sql.impl.JdbcConnection; |
| import org.apache.beam.sdk.extensions.sql.impl.JdbcDriver; |
| import org.apache.beam.sdk.extensions.sql.impl.planner.BeamRuleSets; |
| import org.apache.beam.sdk.extensions.sql.impl.rel.BeamRelNode; |
| import org.apache.beam.sdk.extensions.sql.impl.rel.BeamSqlRelUtils; |
| import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable; |
| import org.apache.beam.sdk.extensions.sql.meta.provider.ReadOnlyTableProvider; |
| import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider; |
| import org.apache.beam.sdk.options.PipelineOptionsFactory; |
| import org.apache.beam.sdk.schemas.Schema; |
| import org.apache.beam.sdk.schemas.Schema.Field; |
| import org.apache.beam.sdk.schemas.Schema.FieldType; |
| import org.apache.beam.sdk.testing.PAssert; |
| import org.apache.beam.sdk.testing.TestPipeline; |
| import org.apache.beam.sdk.values.PCollection; |
| import org.apache.beam.sdk.values.Row; |
| import org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.plan.Context; |
| import org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.plan.Contexts; |
| import org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.plan.ConventionTraitDef; |
| import org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.plan.RelTraitDef; |
| import org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.schema.SchemaPlus; |
| import org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.tools.FrameworkConfig; |
| import org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.tools.Frameworks; |
| import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList; |
| import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap; |
| import org.joda.time.DateTime; |
| import org.joda.time.Duration; |
| import org.joda.time.chrono.ISOChronology; |
| import org.junit.Assert; |
| import org.junit.Before; |
| import org.junit.Ignore; |
| import org.junit.Rule; |
| import org.junit.Test; |
| import org.junit.rules.ExpectedException; |
| import org.junit.runner.RunWith; |
| import org.junit.runners.JUnit4; |
| |
| /** ZetaSQLDialectSpecTest. */ |
| @RunWith(JUnit4.class) |
| public class ZetaSQLDialectSpecTest { |
| private static final Long PIPELINE_EXECUTION_WAITTIME_MINUTES = 2L; |
| |
| private FrameworkConfig config; |
| |
| private TableProvider tableProvider; |
| |
| @Rule public transient TestPipeline pipeline = TestPipeline.create(); |
| @Rule public ExpectedException thrown = ExpectedException.none(); |
| |
| @Before |
| public void setUp() { |
| initializeBeamTableProvider(); |
| initializeCalciteEnvironment(); |
| } |
| |
| @Test |
| public void testSimpleSelect() { |
| String sql = |
| "SELECT CAST (1243 as INT64), " |
| + "CAST ('2018-09-15 12:59:59.000000+00' as TIMESTAMP), " |
| + "CAST ('string' as STRING);"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addDateTimeField("field2") |
| .addStringField("field3") |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues( |
| 1243L, |
| new DateTime(2018, 9, 15, 12, 59, 59, ISOChronology.getInstanceUTC()), |
| "string") |
| .build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testEQ1() { |
| String sql = "SELECT @p0 = @p1 AS ColA"; |
| |
| ImmutableMap<String, Value> params = |
| ImmutableMap.<String, Value>builder() |
| .put("p0", Value.createSimpleNullValue(TypeKind.TYPE_BOOL)) |
| .put("p1", Value.createBoolValue(true)) |
| .build(); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((Boolean) null).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore( |
| "Does not support inf/-inf/nan in double/float literals because double/float literals are" |
| + " converted to BigDecimal in Calcite codegen.") |
| public void testEQ2() { |
| String sql = "SELECT @p0 = @p1 AS ColA"; |
| |
| ImmutableMap<String, Value> params = |
| ImmutableMap.<String, Value>builder() |
| .put("p0", Value.createDoubleValue(0)) |
| .put("p1", Value.createDoubleValue(Double.POSITIVE_INFINITY)) |
| .build(); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addBooleanField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(false).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testEQ3() { |
| String sql = "SELECT @p0 = @p1 AS ColA"; |
| |
| ImmutableMap<String, Value> params = |
| ImmutableMap.<String, Value>builder() |
| .put("p0", Value.createSimpleNullValue(TypeKind.TYPE_DOUBLE)) |
| .put("p1", Value.createDoubleValue(3.14)) |
| .build(); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((Boolean) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testEQ4() { |
| String sql = "SELECT @p0 = @p1 AS ColA"; |
| |
| ImmutableMap<String, Value> params = |
| ImmutableMap.<String, Value>builder() |
| .put("p0", Value.createBytesValue(ByteString.copyFromUtf8("hello"))) |
| .put("p1", Value.createBytesValue(ByteString.copyFromUtf8("hello"))) |
| .build(); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(true).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testEQ5() { |
| String sql = "SELECT b'hello' = b'hello' AS ColA"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(true).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testIsNotNull1() { |
| String sql = "SELECT @p0 IS NOT NULL AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of("p0", Value.createSimpleNullValue(TypeKind.TYPE_STRING)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(false).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testIsNotNull2() { |
| String sql = "SELECT @p0 IS NOT NULL AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createNullValue( |
| TypeFactory.createArrayType(TypeFactory.createSimpleType(TypeKind.TYPE_INT64)))); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(false).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("Does not support struct literal.") |
| public void testIsNotNull3() { |
| String sql = "SELECT @p0 IS NOT NULL AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createNullValue( |
| TypeFactory.createStructType( |
| Arrays.asList( |
| new StructField( |
| "a", TypeFactory.createSimpleType(TypeKind.TYPE_STRING)))))); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(false).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testIfBasic() { |
| String sql = "SELECT IF(@p0, @p1, @p2) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createBoolValue(true), |
| "p1", |
| Value.createInt64Value(1), |
| "p2", |
| Value.createInt64Value(2)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.INT64).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(1L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCoalesceBasic() { |
| String sql = "SELECT COALESCE(@p0, @p1, @p2) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createSimpleNullValue(TypeKind.TYPE_STRING), |
| "p1", |
| Value.createStringValue("yay"), |
| "p2", |
| Value.createStringValue("nay")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.STRING).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("yay").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCoalesceSingleArgument() { |
| String sql = "SELECT COALESCE(@p0) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of("p0", Value.createSimpleNullValue(TypeKind.TYPE_INT64)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder().addNullableField("field1", FieldType.array(FieldType.INT64)).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValue(null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCoalesceNullArray() { |
| String sql = "SELECT COALESCE(@p0, @p1) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createNullValue( |
| TypeFactory.createArrayType(TypeFactory.createSimpleType(TypeKind.TYPE_INT64))), |
| "p1", |
| Value.createNullValue( |
| TypeFactory.createArrayType(TypeFactory.createSimpleType(TypeKind.TYPE_INT64)))); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder().addNullableField("field1", FieldType.array(FieldType.INT64)).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValue(null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testNullIfCoercion() { |
| String sql = "SELECT NULLIF(@p0, @p1) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createInt64Value(3L), |
| "p1", |
| Value.createSimpleNullValue(TypeKind.TYPE_DOUBLE)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.DOUBLE).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValue(3.0).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("Struct literals are not currently supported") |
| public void testCoalesceNullStruct() { |
| String sql = "SELECT COALESCE(NULL, STRUCT(\"a\" AS s, -33 AS i))"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema innerSchema = |
| Schema.of(Field.of("s", FieldType.STRING), Field.of("i", FieldType.INT64)); |
| final Schema schema = |
| Schema.builder().addNullableField("field1", FieldType.row(innerSchema)).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValue(Row.withSchema(innerSchema).addValues("a", -33).build()) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testIfTimestamp() { |
| String sql = "SELECT IF(@p0, @p1, @p2) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createBoolValue(false), |
| "p1", |
| Value.createTimestampValueFromUnixMicros(0), |
| "p2", |
| Value.createTimestampValueFromUnixMicros( |
| DateTime.parse("2019-01-01T00:00:00Z").getMillis() * 1000)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", DATETIME).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(DateTime.parse("2019-01-01T00:00:00Z")).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("$make_array is not implemented") |
| public void testMakeArray() { |
| String sql = "SELECT [s3, s1, s2] FROM (SELECT \"foo\" AS s1, \"bar\" AS s2, \"baz\" AS s3);"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder().addNullableField("field1", FieldType.array(FieldType.STRING)).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValue(ImmutableList.of("baz", "foo", "bar")).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testNullIfPositive() { |
| String sql = "SELECT NULLIF(@p0, @p1) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue("null"), "p1", Value.createStringValue("null")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.STRING).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValue(null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testNullIfNegative() { |
| String sql = "SELECT NULLIF(@p0, @p1) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue("foo"), "p1", Value.createStringValue("null")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.STRING).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("foo").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testIfNullPositive() { |
| String sql = "SELECT IFNULL(@p0, @p1) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue("foo"), "p1", Value.createStringValue("default")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.STRING).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("foo").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testIfNullNegative() { |
| String sql = "SELECT IFNULL(@p0, @p1) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createSimpleNullValue(TypeKind.TYPE_STRING), |
| "p1", |
| Value.createStringValue("yay")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.STRING).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("yay").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("Throws IndexOutOfBoundsException") |
| public void testConstructEmptyArrayLiteral() { |
| String sql = "SELECT @p0 AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createArrayValue( |
| TypeFactory.createArrayType(TypeFactory.createSimpleType(TypeKind.TYPE_INT64)), |
| ImmutableList.of())); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addArrayField("field1", FieldType.INT64).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValue(ImmutableList.of()).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("") |
| public void testLike1() { |
| String sql = "SELECT @p0 LIKE @p1 AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue("ab%"), "p1", Value.createStringValue("ab\\%")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(true).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testLikeNullPattern() { |
| String sql = "SELECT @p0 LIKE @p1 AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createStringValue("ab%"), |
| "p1", |
| Value.createSimpleNullValue(TypeKind.TYPE_STRING)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((Object) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("") |
| public void testLikeAllowsEscapingNonSpecialCharacter() { |
| String sql = "SELECT @p0 LIKE @p1 AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of("p0", Value.createStringValue("ab"), "p1", Value.createStringValue("\\ab")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(true).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("") |
| public void testLikeAllowsEscapingBackslash() { |
| String sql = "SELECT @p0 LIKE @p1 AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue("a\\c"), "p1", Value.createStringValue("a\\\\c")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(true).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("Currently non UTF-8 values are coerced to UTF-8") |
| public void testThrowsErrorForNonUTF8() { |
| String sql = "SELECT @p0 LIKE @p1 AS ColA"; |
| byte[] bytes = {(byte) 0xe8, (byte) 0xb0}; |
| Value bad = |
| Value.deserialize( |
| TypeFactory.createSimpleType(TypeKind.TYPE_STRING), |
| ValueProto.newBuilder().setStringValueBytes(ByteString.copyFrom(bytes)).build()); |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of("p0", Value.createStringValue("abc"), "p1", bad); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| thrown.expect(RuntimeException.class); |
| // TODO: message should be a constant on ZetaSQLPlannerImpl |
| thrown.expectMessage("invalid UTF-8"); |
| zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| } |
| |
| @Test |
| @Ignore("Does not support BYTES for like") |
| public void testLikeBytes() { |
| String sql = "SELECT @p0 LIKE @p1 AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createBytesValue(ByteString.copyFromUtf8("abcd")), |
| "p1", |
| Value.createBytesValue(ByteString.copyFromUtf8("__%"))); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(true).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testMod() { |
| String sql = "SELECT MOD(4, 2)"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(0L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSimpleUnionAll() { |
| String sql = |
| "SELECT CAST (1243 as INT64), " |
| + "CAST ('2018-09-15 12:59:59.000000+00' as TIMESTAMP), " |
| + "CAST ('string' as STRING) " |
| + " UNION ALL " |
| + " SELECT CAST (1243 as INT64), " |
| + "CAST ('2018-09-15 12:59:59.000000+00' as TIMESTAMP), " |
| + "CAST ('string' as STRING);"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addDateTimeField("field2") |
| .addStringField("field3") |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues( |
| 1243L, |
| new DateTime(2018, 9, 15, 12, 59, 59, ISOChronology.getInstanceUTC()), |
| "string") |
| .build(), |
| Row.withSchema(schema) |
| .addValues( |
| 1243L, |
| new DateTime(2018, 9, 15, 12, 59, 59, ISOChronology.getInstanceUTC()), |
| "string") |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testThreeWayUnionAll() { |
| String sql = "SELECT a FROM (SELECT 1 a UNION ALL SELECT 2 UNION ALL SELECT 3)"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(1L).build(), |
| Row.withSchema(schema).addValues(2L).build(), |
| Row.withSchema(schema).addValues(3L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSimpleUnionDISTINCT() { |
| String sql = |
| "SELECT CAST (1243 as INT64), " |
| + "CAST ('2018-09-15 12:59:59.000000+00' as TIMESTAMP), " |
| + "CAST ('string' as STRING) " |
| + " UNION DISTINCT " |
| + " SELECT CAST (1243 as INT64), " |
| + "CAST ('2018-09-15 12:59:59.000000+00' as TIMESTAMP), " |
| + "CAST ('string' as STRING);"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addDateTimeField("field2") |
| .addStringField("field3") |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues( |
| 1243L, |
| new DateTime(2018, 9, 15, 12, 59, 59, ISOChronology.getInstanceUTC()), |
| "string") |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLInnerJoin() { |
| String sql = |
| "SELECT t1.Key " |
| + "FROM KeyValue AS t1" |
| + " INNER JOIN BigTable AS t2" |
| + " on " |
| + " t1.Key = t2.RowKey AND t1.ts = t2.ts"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(Schema.builder().addInt64Field("field1").build()) |
| .addValues(15L) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| // JOIN USING(col) is equivalent to JOIN on left.col = right.col. |
| public void testZetaSQLInnerJoinWithUsing() { |
| String sql = "SELECT t1.Key " + "FROM KeyValue AS t1" + " INNER JOIN BigTable AS t2 USING(ts)"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(Schema.builder().addInt64Field("field1").build()) |
| .addValues(15L) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| // testing ordering of the JOIN conditions. |
| public void testZetaSQLInnerJoinTwo() { |
| String sql = |
| "SELECT t2.RowKey " |
| + "FROM KeyValue AS t1" |
| + " INNER JOIN BigTable AS t2" |
| + " on " |
| + " t2.RowKey = t1.Key AND t2.ts = t1.ts"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(Schema.builder().addInt64Field("field1").build()) |
| .addValues(15L) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLLeftOuterJoin() { |
| String sql = |
| "SELECT * " |
| + "FROM KeyValue AS t1" |
| + " LEFT JOIN BigTable AS t2" |
| + " on " |
| + " t1.Key = t2.RowKey"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schemaOne = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addStringField("field2") |
| .addDateTimeField("field3") |
| .addNullableField("field4", FieldType.INT64) |
| .addNullableField("field5", FieldType.STRING) |
| .addNullableField("field6", DATETIME) |
| .build(); |
| |
| final Schema schemaTwo = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addStringField("field2") |
| .addDateTimeField("field3") |
| .addInt64Field("field4") |
| .addStringField("field5") |
| .addDateTimeField("field6") |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schemaOne) |
| .addValues( |
| 14L, |
| "KeyValue234", |
| new DateTime(2018, 7, 1, 21, 26, 6, ISOChronology.getInstanceUTC()), |
| null, |
| null, |
| null) |
| .build(), |
| Row.withSchema(schemaTwo) |
| .addValues( |
| 15L, |
| "KeyValue235", |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC()), |
| 15L, |
| "BigTable235", |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC())) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLRightOuterJoin() { |
| String sql = |
| "SELECT * " |
| + "FROM KeyValue AS t1" |
| + " RIGHT JOIN BigTable AS t2" |
| + " on " |
| + " t1.Key = t2.RowKey"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schemaOne = |
| Schema.builder() |
| .addNullableField("field1", FieldType.INT64) |
| .addNullableField("field2", FieldType.STRING) |
| .addNullableField("field3", DATETIME) |
| .addInt64Field("field4") |
| .addStringField("field5") |
| .addDateTimeField("field6") |
| .build(); |
| |
| final Schema schemaTwo = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addStringField("field2") |
| .addDateTimeField("field3") |
| .addInt64Field("field4") |
| .addStringField("field5") |
| .addDateTimeField("field6") |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schemaOne) |
| .addValues( |
| null, |
| null, |
| null, |
| 16L, |
| "BigTable236", |
| new DateTime(2018, 7, 1, 21, 26, 8, ISOChronology.getInstanceUTC())) |
| .build(), |
| Row.withSchema(schemaTwo) |
| .addValues( |
| 15L, |
| "KeyValue235", |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC()), |
| 15L, |
| "BigTable235", |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC())) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLFullOuterJoin() { |
| String sql = |
| "SELECT * " |
| + "FROM KeyValue AS t1" |
| + " FULL JOIN BigTable AS t2" |
| + " on " |
| + " t1.Key = t2.RowKey"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schemaOne = |
| Schema.builder() |
| .addNullableField("field1", FieldType.INT64) |
| .addNullableField("field2", FieldType.STRING) |
| .addNullableField("field3", DATETIME) |
| .addInt64Field("field4") |
| .addStringField("field5") |
| .addDateTimeField("field6") |
| .build(); |
| |
| final Schema schemaTwo = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addStringField("field2") |
| .addDateTimeField("field3") |
| .addInt64Field("field4") |
| .addStringField("field5") |
| .addDateTimeField("field6") |
| .build(); |
| |
| final Schema schemaThree = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addStringField("field2") |
| .addDateTimeField("field3") |
| .addNullableField("field4", FieldType.INT64) |
| .addNullableField("field5", FieldType.STRING) |
| .addNullableField("field6", DATETIME) |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schemaOne) |
| .addValues( |
| null, |
| null, |
| null, |
| 16L, |
| "BigTable236", |
| new DateTime(2018, 7, 1, 21, 26, 8, ISOChronology.getInstanceUTC())) |
| .build(), |
| Row.withSchema(schemaTwo) |
| .addValues( |
| 15L, |
| "KeyValue235", |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC()), |
| 15L, |
| "BigTable235", |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC())) |
| .build(), |
| Row.withSchema(schemaThree) |
| .addValues( |
| 14L, |
| "KeyValue234", |
| new DateTime(2018, 7, 1, 21, 26, 6, ISOChronology.getInstanceUTC()), |
| null, |
| null, |
| null) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("BeamSQL only supports equal join") |
| public void testZetaSQLFullOuterJoinTwo() { |
| String sql = |
| "SELECT * " |
| + "FROM KeyValue AS t1" |
| + " FULL JOIN BigTable AS t2" |
| + " on " |
| + " t1.Key + t2.RowKey = 30"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLThreeWayInnerJoin() { |
| String sql = |
| "SELECT t3.Value, t2.Value, t1.Value, t1.Key, t3.ColId FROM KeyValue as t1 " |
| + "JOIN BigTable as t2 " |
| + "ON (t1.Key = t2.RowKey) " |
| + "JOIN Spanner as t3 " |
| + "ON (t3.ColId = t1.Key)"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema( |
| Schema.builder() |
| .addStringField("t3.Value") |
| .addStringField("t2.Value") |
| .addStringField("t1.Value") |
| .addInt64Field("t1.Key") |
| .addInt64Field("t3.ColId") |
| .build()) |
| .addValues("Spanner235", "BigTable235", "KeyValue235", 15L, 15L) |
| .build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLTableJoinOnItselfWithFiltering() { |
| String sql = |
| "SELECT * FROM Spanner as t1 " |
| + "JOIN Spanner as t2 " |
| + "ON (t1.ColId = t2.ColId) WHERE t1.ColId = 17"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema( |
| Schema.builder() |
| .addInt64Field("field1") |
| .addStringField("field2") |
| .addInt64Field("field3") |
| .addStringField("field4") |
| .build()) |
| .addValues(17L, "Spanner237", 17L, "Spanner237") |
| .build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLSelectFromSelect() { |
| String sql = "SELECT * FROM (SELECT \"apple\" AS fruit, \"carrot\" AS vegetable);"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder().addStringField("field1").addStringField("field2").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues("apple", "carrot").build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| |
| Schema outputSchema = stream.getSchema(); |
| Assert.assertEquals(2, outputSchema.getFieldCount()); |
| Assert.assertEquals("fruit", outputSchema.getField(0).getName()); |
| Assert.assertEquals("vegetable", outputSchema.getField(1).getName()); |
| } |
| |
| @Test |
| public void testZetaSQLSelectFromTable() { |
| String sql = "SELECT Key, Value FROM KeyValue;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").addStringField("field2").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(14L, "KeyValue234").build(), |
| Row.withSchema(schema).addValues(15L, "KeyValue235").build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLSelectFromTableLimit() { |
| String sql = "SELECT Key, Value FROM KeyValue LIMIT 2;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").addStringField("field2").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(14L, "KeyValue234").build(), |
| Row.withSchema(schema).addValues(15L, "KeyValue235").build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLSelectFromTableLimit0() { |
| String sql = "SELECT Key, Value FROM KeyValue LIMIT 0;"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| PAssert.that(stream).containsInAnyOrder(); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLSelectFromTableLimitOffset() { |
| String sql = |
| "SELECT COUNT(a) FROM (\n" |
| + "SELECT a FROM (SELECT 1 a UNION ALL SELECT 2 UNION ALL SELECT 3) LIMIT 3 OFFSET 1);"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addInt64Field("field1").build(); |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(2L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| // There is really no order for a PCollection, so this query does not test |
| // ORDER BY but just a test to see if ORDER BY LIMIT can work. |
| @Test |
| public void testZetaSQLSelectFromTableOrderByLimit() { |
| String sql = "SELECT Key, Value FROM KeyValue ORDER BY Key DESC LIMIT 2;"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").addStringField("field2").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(14L, "KeyValue234").build(), |
| Row.withSchema(schema).addValues(15L, "KeyValue235").build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLSelectFromTableOrderBy() { |
| String sql = "SELECT Key, Value FROM KeyValue ORDER BY Key DESC;"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| thrown.expect(RuntimeException.class); |
| thrown.expectMessage("ORDER BY without a LIMIT is not supported."); |
| zetaSQLQueryPlanner.convertToBeamRel(sql); |
| } |
| |
| @Test |
| public void testZetaSQLSelectFromTableWithStructType2() { |
| String sql = |
| "SELECT table_with_struct.struct_col.struct_col_str FROM table_with_struct WHERE id = 1;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addStringField("field").build(); |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValue("row_one").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLStructFieldAccessInFilter() { |
| String sql = |
| "SELECT table_with_struct.id FROM table_with_struct WHERE" |
| + " table_with_struct.struct_col.struct_col_str = 'row_one';"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addInt64Field("field").build(); |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValue(1L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLStructFieldAccessInCast() { |
| String sql = |
| "SELECT CAST(table_with_struct.id AS STRING) FROM table_with_struct WHERE" |
| + " table_with_struct.struct_col.struct_col_str = 'row_one';"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addStringField("field").build(); |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValue("1").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLStructFieldAccessInCast2() { |
| String sql = |
| "SELECT CAST(A.struct_col.struct_col_str AS TIMESTAMP) FROM table_with_struct_ts_string AS" |
| + " A"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addDateTimeField("field").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValue(parseTimestampWithUTCTimeZone("2019-01-15 13:21:03")) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLStructFieldAccessInTumble() { |
| String sql = |
| "SELECT TUMBLE_START('INTERVAL 1 MINUTE') FROM table_with_struct_ts_string AS A GROUP BY " |
| + "TUMBLE(CAST(A.struct_col.struct_col_str AS TIMESTAMP), 'INTERVAL 1 MINUTE')"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addDateTimeField("field").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValue(parseTimestampWithUTCTimeZone("2019-01-15 13:21:00")) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLStructFieldAccessInGroupBy() { |
| String sql = "SELECT rowCol.row_id, COUNT(*) FROM table_with_struct_two GROUP BY rowCol.row_id"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addInt64Field("field1").addInt64Field("field2").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(1L, 1L).build(), |
| Row.withSchema(schema).addValues(2L, 1L).build(), |
| Row.withSchema(schema).addValues(3L, 2L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLStructFieldAccessInGroupBy2() { |
| String sql = |
| "SELECT rowCol.data, MAX(rowCol.row_id), MIN(rowCol.row_id) FROM table_with_struct_two" |
| + " GROUP BY rowCol.data"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = |
| Schema.builder() |
| .addStringField("field1") |
| .addInt64Field("field2") |
| .addInt64Field("field3") |
| .build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues("data1", 1L, 1L).build(), |
| Row.withSchema(schema).addValues("data2", 3L, 2L).build(), |
| Row.withSchema(schema).addValues("data3", 3L, 3L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLSelectFromTableWithArrayType() { |
| String sql = "SELECT array_col FROM table_with_array;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addArrayField("field", FieldType.STRING).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValue(Arrays.asList("1", "2", "3")).build(), |
| Row.withSchema(schema).addValue(ImmutableList.of()).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLSelectStarFromTable() { |
| String sql = "SELECT * FROM BigTable;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addStringField("field2") |
| .addDateTimeField("field3") |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues( |
| 15L, |
| "BigTable235", |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC())) |
| .build(), |
| Row.withSchema(schema) |
| .addValues( |
| 16L, |
| "BigTable236", |
| new DateTime(2018, 7, 1, 21, 26, 8, ISOChronology.getInstanceUTC())) |
| .build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLBasicFiltering() { |
| String sql = "SELECT Key, Value FROM KeyValue WHERE Key = 14;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema( |
| Schema.builder().addInt64Field("field1").addStringField("field2").build()) |
| .addValues(14L, "KeyValue234") |
| .build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLBasicFilteringTwo() { |
| String sql = "SELECT Key, Value FROM KeyValue WHERE Key = 14 AND Value = 'non-existing';"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| PAssert.that(stream).containsInAnyOrder(); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLBasicFilteringThree() { |
| String sql = "SELECT Key, Value FROM KeyValue WHERE Key = 14 OR Key = 15;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").addStringField("field2").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(14L, "KeyValue234").build(), |
| Row.withSchema(schema).addValues(15L, "KeyValue235").build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLCountOnAColumn() { |
| String sql = "SELECT COUNT(Key) FROM KeyValue"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(2L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLAggDistinct() { |
| String sql = "SELECT Key, COUNT(DISTINCT Value) FROM KeyValue GROUP BY Key"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| thrown.expect(RuntimeException.class); |
| thrown.expectMessage("Does not support COUNT DISTINCT"); |
| zetaSQLQueryPlanner.convertToBeamRel(sql); |
| } |
| |
| @Test |
| public void testZetaSQLBasicAgg() { |
| String sql = "SELECT Key, COUNT(*) FROM KeyValue GROUP BY Key"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").addInt64Field("field2").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(14L, 1L).build(), |
| Row.withSchema(schema).addValues(15L, 1L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLColumnAlias1() { |
| String sql = "SELECT Key, COUNT(*) AS count_col FROM KeyValue GROUP BY Key"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| |
| Schema outputSchema = stream.getSchema(); |
| Assert.assertEquals(2, outputSchema.getFieldCount()); |
| Assert.assertEquals("Key", outputSchema.getField(0).getName()); |
| Assert.assertEquals("count_col", outputSchema.getField(1).getName()); |
| } |
| |
| @Test |
| public void testZetaSQLColumnAlias2() { |
| String sql = |
| "SELECT Key AS k1, (count_col + 1) AS k2 FROM (SELECT Key, COUNT(*) AS count_col FROM" |
| + " KeyValue GROUP BY Key)"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| |
| Schema outputSchema = stream.getSchema(); |
| Assert.assertEquals(2, outputSchema.getFieldCount()); |
| Assert.assertEquals("k1", outputSchema.getField(0).getName()); |
| Assert.assertEquals("k2", outputSchema.getField(1).getName()); |
| } |
| |
| @Test |
| public void testZetaSQLColumnAlias3() { |
| String sql = "SELECT Key AS v1, Value AS v2, ts AS v3 FROM KeyValue"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| |
| Schema outputSchema = stream.getSchema(); |
| Assert.assertEquals(3, outputSchema.getFieldCount()); |
| Assert.assertEquals("v1", outputSchema.getField(0).getName()); |
| Assert.assertEquals("v2", outputSchema.getField(1).getName()); |
| Assert.assertEquals("v3", outputSchema.getField(2).getName()); |
| } |
| |
| @Test |
| public void testZetaSQLColumnAlias4() { |
| String sql = "SELECT CAST(123 AS INT64) AS cast_col"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| |
| Schema outputSchema = stream.getSchema(); |
| Assert.assertEquals(1, outputSchema.getFieldCount()); |
| Assert.assertEquals("cast_col", outputSchema.getField(0).getName()); |
| } |
| |
| @Test |
| public void testZetaSQLAmbiguousAlias() { |
| String sql = "SELECT row_id as ID, int64_col as ID FROM table_all_types GROUP BY ID;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| |
| thrown.expectMessage( |
| "Name ID in GROUP BY clause is ambiguous; it may refer to multiple columns in the" |
| + " SELECT-list [at 1:68]"); |
| zetaSQLQueryPlanner.convertToBeamRel(sql); |
| } |
| |
| @Test |
| public void testZetaSQLAggWithOrdinalReference() { |
| String sql = "SELECT Key, COUNT(*) FROM aggregate_test_table GROUP BY 1"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").addInt64Field("field2").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(1L, 2L).build(), |
| Row.withSchema(schema).addValues(2L, 3L).build(), |
| Row.withSchema(schema).addValues(3L, 2L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLAggWithAliasReference() { |
| String sql = "SELECT Key AS K, COUNT(*) FROM aggregate_test_table GROUP BY K"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").addInt64Field("field2").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(1L, 2L).build(), |
| Row.withSchema(schema).addValues(2L, 3L).build(), |
| Row.withSchema(schema).addValues(3L, 2L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLBasicAgg2() { |
| String sql = "SELECT Key, COUNT(*) FROM aggregate_test_table GROUP BY Key"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").addInt64Field("field2").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(1L, 2L).build(), |
| Row.withSchema(schema).addValues(2L, 3L).build(), |
| Row.withSchema(schema).addValues(3L, 2L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLBasicAgg3() { |
| String sql = "SELECT Key, Key2, COUNT(*) FROM aggregate_test_table GROUP BY Key2, Key"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addInt64Field("field3") |
| .addInt64Field("field2") |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(1L, 10L, 1L).build(), |
| Row.withSchema(schema).addValues(1L, 11L, 1L).build(), |
| Row.withSchema(schema).addValues(2L, 11L, 2L).build(), |
| Row.withSchema(schema).addValues(2L, 12L, 1L).build(), |
| Row.withSchema(schema).addValues(3L, 13L, 2L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLBasicAgg4() { |
| String sql = |
| "SELECT Key, Key2, MAX(f_int_1), MIN(f_int_1), SUM(f_int_1), SUM(f_double_1) " |
| + "FROM aggregate_test_table GROUP BY Key2, Key"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addInt64Field("field3") |
| .addInt64Field("field2") |
| .addInt64Field("field4") |
| .addInt64Field("field5") |
| .addDoubleField("field6") |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(1L, 10L, 1L, 1L, 1L, 1.0).build(), |
| Row.withSchema(schema).addValues(1L, 11L, 2L, 2L, 2L, 2.0).build(), |
| Row.withSchema(schema).addValues(2L, 11L, 4L, 3L, 7L, 7.0).build(), |
| Row.withSchema(schema).addValues(2L, 12L, 5L, 5L, 5L, 5.0).build(), |
| Row.withSchema(schema).addValues(3L, 13L, 7L, 6L, 13L, 13.0).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLBasicAgg5() { |
| String sql = |
| "SELECT Key, Key2, AVG(CAST(f_int_1 AS FLOAT64)), AVG(f_double_1) " |
| + "FROM aggregate_test_table GROUP BY Key2, Key"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addInt64Field("field2") |
| .addDoubleField("field3") |
| .addDoubleField("field4") |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(1L, 10L, 1.0, 1.0).build(), |
| Row.withSchema(schema).addValues(1L, 11L, 2.0, 2.0).build(), |
| Row.withSchema(schema).addValues(2L, 11L, 3.5, 3.5).build(), |
| Row.withSchema(schema).addValues(2L, 12L, 5.0, 5.0).build(), |
| Row.withSchema(schema).addValues(3L, 13L, 6.5, 6.5).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore( |
| "Calcite infers return type of AVG(int64) as BIGINT while ZetaSQL requires it as either" |
| + " NUMERIC or DOUBLE/FLOAT64") |
| public void testZetaSQLTestAVG() { |
| String sql = "SELECT Key, AVG(f_int_1)" + "FROM aggregate_test_table GROUP BY Key"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addInt64Field("field2") |
| .addInt64Field("field3") |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(1L, 10L, 1L).build(), |
| Row.withSchema(schema).addValues(1L, 11L, 6L).build(), |
| Row.withSchema(schema).addValues(2L, 11L, 6L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLGroupByExprInSelect() { |
| String sql = "SELECT int64_col + 1 FROM table_all_types GROUP BY int64_col + 1;"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValue(0L).build(), |
| Row.withSchema(schema).addValue(-1L).build(), |
| Row.withSchema(schema).addValue(-2L).build(), |
| Row.withSchema(schema).addValue(-3L).build(), |
| Row.withSchema(schema).addValue(-4L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLGroupByAndFiltering() { |
| String sql = "SELECT int64_col FROM table_all_types WHERE int64_col = 1 GROUP BY int64_col;"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| PAssert.that(stream).containsInAnyOrder(); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLGroupByAndFilteringOnNonGroupByColumn() { |
| String sql = "SELECT int64_col FROM table_all_types WHERE double_col = 0.5 GROUP BY int64_col;"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addInt64Field("field").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValue(-5L).build(), |
| Row.withSchema(schema).addValue(-4L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLBasicHaving() { |
| String sql = "SELECT Key, COUNT(*) FROM aggregate_test_table GROUP BY Key HAVING COUNT(*) > 2"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").addInt64Field("field2").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(2L, 3L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLBasicFixedWindowing() { |
| String sql = |
| "SELECT " |
| + "COUNT(*) as field_count, " |
| + "TUMBLE_START(\"INTERVAL 1 SECOND\") as window_start, " |
| + "TUMBLE_END(\"INTERVAL 1 SECOND\") as window_end " |
| + "FROM KeyValue " |
| + "GROUP BY TUMBLE(ts, \"INTERVAL 1 SECOND\");"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("count_start") |
| .addDateTimeField("field1") |
| .addDateTimeField("field2") |
| .build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues( |
| 1L, |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 8, ISOChronology.getInstanceUTC())) |
| .build(), |
| Row.withSchema(schema) |
| .addValues( |
| 1L, |
| new DateTime(2018, 7, 1, 21, 26, 6, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC())) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLBasicSlidingWindowing() { |
| String sql = |
| "SELECT " |
| + "COUNT(*) as field_count, " |
| + "HOP_START(\"INTERVAL 1 SECOND\", \"INTERVAL 2 SECOND\") as window_start, " |
| + "HOP_END(\"INTERVAL 1 SECOND\", \"INTERVAL 2 SECOND\") as window_end " |
| + "FROM window_test_table " |
| + "GROUP BY HOP(ts, \"INTERVAL 1 SECOND\", \"INTERVAL 2 SECOND\");"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("count_star") |
| .addDateTimeField("field1") |
| .addDateTimeField("field2") |
| .build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues( |
| 2L, |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 9, ISOChronology.getInstanceUTC())) |
| .build(), |
| Row.withSchema(schema) |
| .addValues( |
| 1L, |
| new DateTime(2018, 7, 1, 21, 26, 5, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC())) |
| .build(), |
| Row.withSchema(schema) |
| .addValues( |
| 2L, |
| new DateTime(2018, 7, 1, 21, 26, 6, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 8, ISOChronology.getInstanceUTC())) |
| .build(), |
| Row.withSchema(schema) |
| .addValues( |
| 2L, |
| new DateTime(2018, 7, 1, 21, 26, 8, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 10, ISOChronology.getInstanceUTC())) |
| .build(), |
| Row.withSchema(schema) |
| .addValues( |
| 1L, |
| new DateTime(2018, 7, 1, 21, 26, 9, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 11, ISOChronology.getInstanceUTC())) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLBasicSessionWindowing() { |
| String sql = |
| "SELECT " |
| + "COUNT(*) as field_count, " |
| + "SESSION_START(\"INTERVAL 3 SECOND\") as window_start, " |
| + "SESSION_END(\"INTERVAL 3 SECOND\") as window_end " |
| + "FROM window_test_table_two " |
| + "GROUP BY SESSION(ts, \"INTERVAL 3 SECOND\");"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("count_star") |
| .addDateTimeField("field1") |
| .addDateTimeField("field2") |
| .build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues( |
| 2L, |
| new DateTime(2018, 7, 1, 21, 26, 12, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 12, ISOChronology.getInstanceUTC())) |
| .build(), |
| Row.withSchema(schema) |
| .addValues( |
| 2L, |
| new DateTime(2018, 7, 1, 21, 26, 6, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 6, ISOChronology.getInstanceUTC())) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| // Test nested selection |
| @Test |
| public void testZetaSQLNestedQueryOne() { |
| String sql = |
| "SELECT a.Value, a.Key FROM (SELECT Key, Value FROM KeyValue WHERE Key = 14 OR Key = 15)" |
| + " as a;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field2").addInt64Field("field1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues("KeyValue234", 14L).build(), |
| Row.withSchema(schema).addValues("KeyValue235", 15L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| // Test selection, filtering and aggregation combined query. |
| @Test |
| public void testZetaSQLNestedQueryTwo() { |
| String sql = |
| "SELECT a.Key, a.Key2, COUNT(*) FROM " |
| + " (SELECT * FROM aggregate_test_table WHERE Key != 10) as a " |
| + " GROUP BY a.Key2, a.Key"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addInt64Field("field3") |
| .addInt64Field("field2") |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(1L, 10L, 1L).build(), |
| Row.withSchema(schema).addValues(1L, 11L, 1L).build(), |
| Row.withSchema(schema).addValues(2L, 11L, 2L).build(), |
| Row.withSchema(schema).addValues(2L, 12L, 1L).build(), |
| Row.withSchema(schema).addValues(3L, 13L, 2L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| // test selection and join combined query |
| @Test |
| public void testZetaSQLNestedQueryThree() { |
| String sql = |
| "SELECT * FROM (SELECT * FROM KeyValue) AS t1 INNER JOIN (SELECT * FROM BigTable) AS t2 on" |
| + " t1.Key = t2.RowKey"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema( |
| Schema.builder() |
| .addInt64Field("Key") |
| .addStringField("Value") |
| .addDateTimeField("ts") |
| .addInt64Field("RowKey") |
| .addStringField("Value2") |
| .addDateTimeField("ts2") |
| .build()) |
| .addValues( |
| 15L, |
| "KeyValue235", |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC()), |
| 15L, |
| "BigTable235", |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC())) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testZetaSQLNestedQueryFour() { |
| String sql = |
| "SELECT t1.Value, TUMBLE_START('INTERVAL 1 SECOND') AS period_start, MIN(t2.Value) as" |
| + " min_v FROM KeyValue AS t1 INNER JOIN BigTable AS t2 on t1.Key = t2.RowKey GROUP BY" |
| + " t1.Value, TUMBLE(t2.ts, 'INTERVAL 1 SECOND')"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema( |
| Schema.builder() |
| .addStringField("value") |
| .addDateTimeField("min_v") |
| .addStringField("period_start") |
| .build()) |
| .addValues( |
| "KeyValue235", |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC()), |
| "BigTable235") |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| // Test nested select with out of order columns. |
| @Test |
| public void testZetaSQLNestedQueryFive() { |
| String sql = |
| "SELECT a.Value, a.Key FROM (SELECT Value, Key FROM KeyValue WHERE Key = 14 OR Key = 15)" |
| + " as a;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field2").addInt64Field("field1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues("KeyValue234", 14L).build(), |
| Row.withSchema(schema).addValues("KeyValue235", 15L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("Does not support DATE_ADD and TIME_ADD.") |
| public void testDateAndTimeAddSub() { |
| String sql = |
| "SELECT " |
| + "DATE_ADD(DATE '2008-12-25', INTERVAL 5 DAY), " |
| + "TIME_ADD(TIME '13:24:30', INTERVAL 3 HOUR)"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema( |
| Schema.builder() |
| .addDateTimeField("f_date_plus") |
| .addDateTimeField("f_time_plus") |
| .build()) |
| .addValues(parseDate("2008-12-30"), parseTime("16:24:30")) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testTimestampAddSub() { |
| String sql = |
| "SELECT " |
| + "TIMESTAMP_ADD(TIMESTAMP '2008-12-25 15:30:00 UTC', INTERVAL 10 MINUTE), " |
| + "TIMESTAMP_ADD(TIMESTAMP '2008-12-25 15:30:00+07:30', INTERVAL 10 MINUTE)"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema( |
| Schema.builder() |
| .addDateTimeField("f_timestamp_plus") |
| .addDateTimeField("f_timestamp_with_time_zone_plus") |
| .build()) |
| .addValues( |
| DateTimeUtils.parseTimestampWithUTCTimeZone("2008-12-25 15:40:00"), |
| parseTimestampWithTimeZone("2008-12-25 15:40:00+0730")) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testTimeZone() { |
| String sql = "SELECT TIMESTAMP '2018-12-10 10:38:59-10:00'"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(Schema.builder().addDateTimeField("f_timestamp_with_time_zone").build()) |
| .addValues(parseTimestampWithTimeZone("2018-12-10 10:38:59-1000")) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testDistinct() { |
| String sql = "SELECT DISTINCT Key2 FROM aggregate_test_table"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| Schema schema = Schema.builder().addInt64Field("Key2").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(10L).build(), |
| Row.withSchema(schema).addValues(11L).build(), |
| Row.withSchema(schema).addValues(12L).build(), |
| Row.withSchema(schema).addValues(13L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testDistinctOnNull() { |
| String sql = "SELECT DISTINCT str_val FROM all_null_table"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| Schema schema = Schema.builder().addNullableField("str_val", FieldType.DOUBLE).build(); |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((Object) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("BeamSQL does not support ANY_VALUE") |
| public void testAnyValue() { |
| String sql = "SELECT ANY_VALUE(double_val) FROM all_null_table"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| Schema schema = Schema.builder().addNullableField("double_val", FieldType.DOUBLE).build(); |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((Object) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSelectNULL() { |
| String sql = "SELECT NULL"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| Schema schema = Schema.builder().addNullableField("long_val", FieldType.INT64).build(); |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((Object) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testWithQueryOne() { |
| String sql = |
| "With T1 AS (SELECT * FROM KeyValue), T2 AS (SELECT * FROM BigTable) SELECT T2.RowKey FROM" |
| + " T1 INNER JOIN T2 on T1.Key = T2.RowKey;"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(Schema.builder().addInt64Field("field1").build()) |
| .addValues(15L) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testWithQueryTwo() { |
| String sql = |
| "WITH T1 AS (SELECT Key, COUNT(*) as value FROM KeyValue GROUP BY Key) SELECT T1.Key," |
| + " T1.value FROM T1"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").addInt64Field("field2").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(14L, 1L).build(), |
| Row.withSchema(schema).addValues(15L, 1L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testWithQueryThree() { |
| String sql = |
| "WITH T1 as (SELECT Value, Key FROM KeyValue WHERE Key = 14 OR Key = 15) SELECT T1.Value," |
| + " T1.Key FROM T1;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").addInt64Field("field2").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues("KeyValue234", 14L).build(), |
| Row.withSchema(schema).addValues("KeyValue235", 15L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testWithQueryFour() { |
| String sql = |
| "WITH T1 as (SELECT Value, Key FROM KeyValue) SELECT T1.Value, T1.Key FROM T1 WHERE T1.Key" |
| + " = 14 OR T1.Key = 15;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field2").addInt64Field("field1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues("KeyValue234", 14L).build(), |
| Row.withSchema(schema).addValues("KeyValue235", 15L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testWithQueryFive() { |
| String sql = |
| "WITH T1 AS (SELECT * FROM KeyValue) SELECT T1.Key, COUNT(*) FROM T1 GROUP BY T1.Key"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").addInt64Field("field2").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(14L, 1L).build(), |
| Row.withSchema(schema).addValues(15L, 1L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testWithQuerySix() { |
| String sql = |
| "WITH T1 AS (SELECT * FROM window_test_table_two) SELECT " |
| + "COUNT(*) as field_count, " |
| + "SESSION_START(\"INTERVAL 3 SECOND\") as window_start, " |
| + "SESSION_END(\"INTERVAL 3 SECOND\") as window_end " |
| + "FROM T1 " |
| + "GROUP BY SESSION(ts, \"INTERVAL 3 SECOND\");"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("count_star") |
| .addDateTimeField("field1") |
| .addDateTimeField("field2") |
| .build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues( |
| 2L, |
| new DateTime(2018, 7, 1, 21, 26, 12, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 12, ISOChronology.getInstanceUTC())) |
| .build(), |
| Row.withSchema(schema) |
| .addValues( |
| 2L, |
| new DateTime(2018, 7, 1, 21, 26, 6, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 6, ISOChronology.getInstanceUTC())) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testWithQuerySeven() { |
| String sql = |
| "WITH T1 AS (SELECT * FROM KeyValue) SELECT " |
| + "COUNT(*) as field_count, " |
| + "TUMBLE_START(\"INTERVAL 1 SECOND\") as window_start, " |
| + "TUMBLE_END(\"INTERVAL 1 SECOND\") as window_end " |
| + "FROM T1 " |
| + "GROUP BY TUMBLE(ts, \"INTERVAL 1 SECOND\");"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("count_start") |
| .addDateTimeField("field1") |
| .addDateTimeField("field2") |
| .build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues( |
| 1L, |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 8, ISOChronology.getInstanceUTC())) |
| .build(), |
| Row.withSchema(schema) |
| .addValues( |
| 1L, |
| new DateTime(2018, 7, 1, 21, 26, 6, ISOChronology.getInstanceUTC()), |
| new DateTime(2018, 7, 1, 21, 26, 7, ISOChronology.getInstanceUTC())) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testUNNESTLiteral() { |
| String sql = "SELECT * FROM UNNEST(ARRAY<STRING>['foo', 'bar']);"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| Schema schema = Schema.builder().addStringField("str_field").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues("foo").build(), |
| Row.withSchema(schema).addValues("bar").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testNamedUNNESTLiteral() { |
| String sql = "SELECT *, T1 FROM UNNEST(ARRAY<STRING>['foo', 'bar']) AS T1"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| Schema schema = |
| Schema.builder().addStringField("str_field").addStringField("str2_field").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues("foo", "foo").build(), |
| Row.withSchema(schema).addValues("bar", "bar").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("Seeing exception in Beam, need further investigation on the cause of this failed query.") |
| public void testNamedUNNESTJoin() { |
| String sql = |
| "SELECT * " |
| + "FROM table_with_array_for_unnest AS t1" |
| + " LEFT JOIN UNNEST(t1.int_array_col) AS t2" |
| + " on " |
| + " t1.int_col = t2"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream).containsInAnyOrder(); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCaseNoValue() { |
| String sql = "SELECT CASE WHEN 1 > 2 THEN 'not possible' ELSE 'seems right' END"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(Schema.builder().addStringField("str_field").build()) |
| .addValue("seems right") |
| .build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCaseWithValue() { |
| String sql = "SELECT CASE 1 WHEN 2 THEN 'not possible' ELSE 'seems right' END"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(Schema.builder().addStringField("str_field").build()) |
| .addValue("seems right") |
| .build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCaseWithValueMultipleCases() { |
| String sql = |
| "SELECT CASE 2 WHEN 1 THEN 'not possible' WHEN 2 THEN 'seems right' ELSE 'also not" |
| + " possible' END"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(Schema.builder().addStringField("str_field").build()) |
| .addValue("seems right") |
| .build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCaseWithValueNoElse() { |
| String sql = "SELECT CASE 2 WHEN 1 THEN 'not possible' WHEN 2 THEN 'seems right' END"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(Schema.builder().addStringField("str_field").build()) |
| .addValue("seems right") |
| .build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCaseNoValueNoElseNoMatch() { |
| String sql = "SELECT CASE WHEN 'abc' = '123' THEN 'not possible' END"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(Schema.builder().addNullableField("str_field", FieldType.STRING).build()) |
| .addValue(null) |
| .build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCaseWithValueNoElseNoMatch() { |
| String sql = "SELECT CASE 2 WHEN 1 THEN 'not possible' END"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(Schema.builder().addNullableField("str_field", FieldType.STRING).build()) |
| .addValue(null) |
| .build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore( |
| "Codegen generates code that Janino cannot compile, need further investigation on root" |
| + " cause.") |
| public void testCastToDateWithCase() { |
| String sql = |
| "SELECT f_int, \n" |
| + "CASE WHEN CHAR_LENGTH(TRIM(f_string)) = 8 \n" |
| + " THEN CAST (CONCAT(\n" |
| + " SUBSTR(TRIM(f_string), 0, 4) \n" |
| + " , '-' \n" |
| + " , SUBSTR(TRIM(f_string), 4, 2) \n" |
| + " , '-' \n" |
| + " , SUBSTR(TRIM(f_string), 6, 2)) AS DATE)\n" |
| + " ELSE NULL\n" |
| + "END \n" |
| + "FROM table_for_case_when"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| Schema resultType = |
| Schema.builder().addInt32Field("f_int").addNullableField("f_date", DATETIME).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(resultType).addValues(1, parseDate("2018-10-18")).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testIntersectAll() { |
| String sql = |
| "SELECT Key FROM aggregate_test_table " |
| + "INTERSECT ALL " |
| + "SELECT Key FROM aggregate_test_table_two"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| Schema resultType = Schema.builder().addInt64Field("field").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(resultType).addValues(1L).build(), |
| Row.withSchema(resultType).addValues(2L).build(), |
| Row.withSchema(resultType).addValues(2L).build(), |
| Row.withSchema(resultType).addValues(2L).build(), |
| Row.withSchema(resultType).addValues(3L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testIntersectDistinct() { |
| String sql = |
| "SELECT Key FROM aggregate_test_table " |
| + "INTERSECT DISTINCT " |
| + "SELECT Key FROM aggregate_test_table_two"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| Schema resultType = Schema.builder().addInt64Field("field").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(resultType).addValues(1L).build(), |
| Row.withSchema(resultType).addValues(2L).build(), |
| Row.withSchema(resultType).addValues(3L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testExceptAll() { |
| String sql = |
| "SELECT Key FROM aggregate_test_table " |
| + "EXCEPT ALL " |
| + "SELECT Key FROM aggregate_test_table_two"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| Schema resultType = Schema.builder().addInt64Field("field").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(resultType).addValues(1L).build(), |
| Row.withSchema(resultType).addValues(3L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSelectFromEmptyTable() { |
| String sql = "SELECT * FROM table_empty;"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| PAssert.that(stream).containsInAnyOrder(); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testStartsWithString() { |
| String sql = "SELECT STARTS_WITH('string1', 'stri')"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(true).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testStartsWithString2() { |
| String sql = "SELECT STARTS_WITH(@p0, @p1)"; |
| |
| ImmutableMap<String, Value> params = |
| ImmutableMap.<String, Value>builder() |
| .put("p0", Value.createSimpleNullValue(TypeKind.TYPE_STRING)) |
| .put("p1", Value.createStringValue("")) |
| .build(); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((Boolean) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testStartsWithString3() { |
| String sql = "SELECT STARTS_WITH(@p0, @p1)"; |
| |
| ImmutableMap<String, Value> params = |
| ImmutableMap.<String, Value>builder() |
| .put("p0", Value.createSimpleNullValue(TypeKind.TYPE_STRING)) |
| .put("p1", Value.createSimpleNullValue(TypeKind.TYPE_STRING)) |
| .build(); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((Boolean) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testEndsWithString() { |
| String sql = "SELECT STARTS_WITH('string1', 'ng0')"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(false).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testEndsWithString2() { |
| String sql = "SELECT STARTS_WITH(@p0, @p1)"; |
| |
| ImmutableMap<String, Value> params = |
| ImmutableMap.<String, Value>builder() |
| .put("p0", Value.createSimpleNullValue(TypeKind.TYPE_STRING)) |
| .put("p1", Value.createStringValue("")) |
| .build(); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((Boolean) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testEndsWithString3() { |
| String sql = "SELECT STARTS_WITH(@p0, @p1)"; |
| |
| ImmutableMap<String, Value> params = |
| ImmutableMap.<String, Value>builder() |
| .put("p0", Value.createSimpleNullValue(TypeKind.TYPE_STRING)) |
| .put("p1", Value.createSimpleNullValue(TypeKind.TYPE_STRING)) |
| .build(); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.BOOLEAN).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((Boolean) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("Does not support DateTime literal.") |
| public void testDateTimeLiteral() { |
| String sql = "SELECT DATETIME '2018-01-01 05:30:00.334'"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| thrown.expect(RuntimeException.class); |
| thrown.expectMessage("Unsupported ResolvedLiteral type: DATETIME"); |
| zetaSQLQueryPlanner.convertToBeamRel(sql); |
| } |
| |
| @Test |
| public void testTimeStampLiteral() { |
| String sql = "SELECT TIMESTAMP '2016-12-25 05:30:00+00'"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addDateTimeField("field1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues(parseTimestampWithUTCTimeZone("2016-12-25 05:30:00")) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testTimeStampLiteralWithoutTimeZone() { |
| String sql = "SELECT TIMESTAMP '2016-12-25 05:30:00'"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addDateTimeField("field1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues(parseTimestampWithUTCTimeZone("2016-12-25 05:30:00")) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testTimeStampLiteralWithNonUTCTimeZone() { |
| String sql = "SELECT TIMESTAMP '2016-12-25 05:30:00+05'"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addDateTimeField("field1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues(parseTimestampWithTimeZone("2016-12-25 05:30:00+05")) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testConcatWithOneParameters() { |
| String sql = "SELECT concat('abc')"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("abc").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testConcatWithTwoParameters() { |
| String sql = "SELECT concat('abc', 'def')"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("abcdef").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testConcatWithThreeParameters() { |
| String sql = "SELECT concat('abc', 'def', 'xyz')"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("abcdefxyz").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testConcatWithFourParameters() { |
| String sql = "SELECT concat('abc', 'def', ' ', 'xyz')"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues("abcdef xyz").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testConcatWithFiveParameters() { |
| String sql = "SELECT concat('abc', 'def', ' ', 'xyz', 'kkk')"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues("abcdef xyzkkk").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore( |
| "Calcite codegen does not support UDF with ... args." |
| + " See:https://jira.apache.org/jira/browse/CALCITE-2889") |
| public void testConcatWithSixParameters() { |
| String sql = "SELECT concat('abc', 'def', ' ', 'xyz', 'kkk', 'ttt')"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues("abcdef xyzkkkttt").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testConcatWithNull1() { |
| String sql = "SELECT CONCAT(@p0, @p1) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createStringValue(""), |
| "p1", |
| Value.createSimpleNullValue(TypeKind.TYPE_STRING)); |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.STRING).build(); |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((String) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testConcatWithNull2() { |
| String sql = "SELECT CONCAT(@p0, @p1) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", |
| Value.createSimpleNullValue(TypeKind.TYPE_STRING), |
| "p1", |
| Value.createSimpleNullValue(TypeKind.TYPE_STRING)); |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.STRING).build(); |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((String) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testConcatParameterQuery() { |
| String sql = "SELECT CONCAT(@p0, @p1) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of("p0", Value.createStringValue(""), "p1", Value.createStringValue("A")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("A").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testReplace1() { |
| String sql = "SELECT REPLACE(@p0, @p1, @p2) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue(""), |
| "p1", Value.createStringValue(""), |
| "p2", Value.createStringValue("a")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testReplace2() { |
| String sql = "SELECT REPLACE(@p0, @p1, @p2) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue("abc"), |
| "p1", Value.createStringValue(""), |
| "p2", Value.createStringValue("xyz")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("abc").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testReplace3() { |
| String sql = "SELECT REPLACE(@p0, @p1, @p2) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue(""), |
| "p1", Value.createStringValue(""), |
| "p2", Value.createSimpleNullValue(TypeKind.TYPE_STRING)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.STRING).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((String) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testReplace4() { |
| String sql = "SELECT REPLACE(@p0, @p1, @p2) AS ColA"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createSimpleNullValue(TypeKind.TYPE_STRING), |
| "p1", Value.createSimpleNullValue(TypeKind.TYPE_STRING), |
| "p2", Value.createStringValue("")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.STRING).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((String) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testTrim1() { |
| String sql = "SELECT trim(@p0)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of("p0", Value.createStringValue(" a b c ")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("a b c").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testTrim2() { |
| String sql = "SELECT trim(@p0, @p1)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue("abxyzab"), "p1", Value.createStringValue("ab")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("xyz").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testTrim3() { |
| String sql = "SELECT trim(@p0, @p1)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createSimpleNullValue(TypeKind.TYPE_STRING), |
| "p1", Value.createSimpleNullValue(TypeKind.TYPE_STRING)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.STRING).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((String) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testLTrim1() { |
| String sql = "SELECT ltrim(@p0)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of("p0", Value.createStringValue(" a b c ")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("a b c ").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testLTrim2() { |
| String sql = "SELECT ltrim(@p0, @p1)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue("abxyzab"), "p1", Value.createStringValue("ab")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("xyzab").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testLTrim3() { |
| String sql = "SELECT ltrim(@p0, @p1)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createSimpleNullValue(TypeKind.TYPE_STRING), |
| "p1", Value.createSimpleNullValue(TypeKind.TYPE_STRING)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.STRING).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((String) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testRTrim1() { |
| String sql = "SELECT rtrim(@p0)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of("p0", Value.createStringValue(" a b c ")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(" a b c").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testRTrim2() { |
| String sql = "SELECT rtrim(@p0, @p1)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue("abxyzab"), "p1", Value.createStringValue("ab")); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("abxyz").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testRTrim3() { |
| String sql = "SELECT rtrim(@p0, @p1)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createSimpleNullValue(TypeKind.TYPE_STRING), |
| "p1", Value.createSimpleNullValue(TypeKind.TYPE_STRING)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field1", FieldType.STRING).build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((String) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("") |
| public void testCastBytesToString1() { |
| String sql = "SELECT CAST(@p0 AS STRING)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of("p0", Value.createBytesValue(ByteString.copyFromUtf8("`"))); |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("`").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCastBytesToString2() { |
| String sql = "SELECT CAST(b'b' AS STRING)"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("b").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("") |
| public void testCastBytesToStringFromTable() { |
| String sql = "SELECT CAST(bytes_col AS STRING) FROM table_all_types"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues("1").build(), |
| Row.withSchema(schema).addValues("2").build(), |
| Row.withSchema(schema).addValues("3").build(), |
| Row.withSchema(schema).addValues("4").build(), |
| Row.withSchema(schema).addValues("5").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCastStringToTS() { |
| String sql = "SELECT CAST('2019-01-15 13:21:03' AS TIMESTAMP)"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addDateTimeField("field_1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues(parseTimestampWithUTCTimeZone("2019-01-15 13:21:03")) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCastStringToString() { |
| String sql = "SELECT CAST(@p0 AS STRING)"; |
| ImmutableMap<String, Value> params = ImmutableMap.of("p0", Value.createStringValue("")); |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCastStringToInt64() { |
| String sql = "SELECT CAST(@p0 AS INT64)"; |
| ImmutableMap<String, Value> params = ImmutableMap.of("p0", Value.createStringValue("123")); |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(123L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSelectConstant() { |
| String sql = "SELECT 'hi'"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("hi").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("Does not support DATE_ADD.") |
| public void testDateAddWithParameter() { |
| String sql = |
| "SELECT " |
| + "DATE_ADD(@p0, INTERVAL @p1 DAY), " |
| + "DATE_ADD(@p2, INTERVAL @p3 DAY), " |
| + "DATE_ADD(@p4, INTERVAL @p5 YEAR), " |
| + "DATE_ADD(@p6, INTERVAL @p7 DAY), " |
| + "DATE_ADD(@p8, INTERVAL @p9 MONTH)"; |
| // Value |
| ImmutableMap<String, Value> params = |
| ImmutableMap.<String, Value>builder() |
| .put("p0", Value.createDateValue(0)) // 1970-01-01 |
| .put("p1", Value.createInt64Value(2L)) |
| .put("p2", parseDateToValue("2019-01-01")) |
| .put("p3", Value.createInt64Value(2L)) |
| .put("p4", Value.createSimpleNullValue(TypeKind.TYPE_DATE)) |
| .put("p5", Value.createInt64Value(1L)) |
| .put("p6", parseDateToValue("2000-02-29")) |
| .put("p7", Value.createInt64Value(-365L)) |
| .put("p8", parseDateToValue("1999-03-31")) |
| .put("p9", Value.createInt64Value(-1L)) |
| .build(); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addDateTimeField("field1") |
| .addDateTimeField("field2") |
| .addNullableField("field3", DATETIME) |
| .addDateTimeField("field4") |
| .addDateTimeField("field5") |
| .build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues( |
| parseDate("1970-01-03"), |
| parseDate("2019-01-03"), |
| null, |
| parseDate("1999-03-01"), |
| parseDate("1999-02-28")) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("Does not support TIME_ADD.") |
| public void testTimeAddWithParameter() { |
| String sql = "SELECT TIME_ADD(@p0, INTERVAL @p1 SECOND)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", parseTimeToValue("12:13:14.123"), |
| "p1", Value.createInt64Value(1L)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addDateTimeField("field1").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues(parseTime("12:13:15.123")).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("") |
| public void testTimestampAddWithParameter() { |
| String sql = "SELECT TIMESTAMP_ADD(@p0, INTERVAL @p1 MILLISECOND)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", parseTimestampWithTZToValue("2001-01-01 00:00:00+00"), |
| "p1", Value.createInt64Value(1L)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addDateTimeField("field1").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues(parseTimestampWithTimeZone("2001-01-01 00:00:00.001+00")) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testTimeStampAddWithParameter() { |
| String sql = "SELECT TIMESTAMP_ADD(@p0, INTERVAL @p1 MINUTE)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", parseTimestampWithTZToValue("2008-12-25 15:30:00+07:30"), |
| "p1", Value.createInt64Value(10L)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addDateTimeField("field1").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues(parseTimestampWithTimeZone("2008-12-25 15:40:00+07:30")) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSelectFromTableWithMap() { |
| String sql = "SELECT row_field FROM table_with_map"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| Schema rowSchema = Schema.builder().addInt64Field("row_id").addStringField("data").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(Schema.builder().addRowField("row_field", rowSchema).build()) |
| .addValues(Row.withSchema(rowSchema).addValues(1L, "data1").build()) |
| .build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSubQuery() { |
| String sql = "select sum(Key) from KeyValue\n" + "group by (select Key)"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| thrown.expect(IllegalArgumentException.class); |
| thrown.expectMessage("Does not support sub-queries"); |
| zetaSQLQueryPlanner.convertToBeamRel(sql); |
| } |
| |
| @Test |
| public void testSubstr() { |
| String sql = "SELECT substr(@p0, @p1, @p2)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue("abc"), |
| "p1", Value.createInt64Value(-2L), |
| "p2", Value.createInt64Value(1L)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field1").build(); |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("b").build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSubstrWithLargeValueExpectException() { |
| String sql = "SELECT substr(@p0, @p1, @p2)"; |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of( |
| "p0", Value.createStringValue("abc"), |
| "p1", Value.createInt64Value(Integer.MAX_VALUE + 1L), |
| "p2", Value.createInt64Value(Integer.MIN_VALUE - 1L)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| thrown.expect(RuntimeException.class); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSelectAll() { |
| String sql = "SELECT ALL Key, Value FROM KeyValue;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").addStringField("field2").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(14L, "KeyValue234").build(), |
| Row.withSchema(schema).addValues(15L, "KeyValue235").build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSelectDistinct() { |
| String sql = "SELECT DISTINCT Key FROM aggregate_test_table;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(1L).build(), |
| Row.withSchema(schema).addValues(2L).build(), |
| Row.withSchema(schema).addValues(3L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("Bytes cannot be in UNION ALL") |
| public void testSelectDistinct2() { |
| String sql = |
| "SELECT DISTINCT val.BYTES\n" |
| + "from (select b\"BYTES\" BYTES union all\n" |
| + " select b\"bytes\" union all\n" |
| + " select b\"ByTeS\") val"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addByteArrayField("field1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues("BYTES".getBytes(StandardCharsets.UTF_8)).build(), |
| Row.withSchema(schema).addValues("ByTeS".getBytes(StandardCharsets.UTF_8)).build(), |
| Row.withSchema(schema).addValues("bytes".getBytes(StandardCharsets.UTF_8)).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSelectBytes() { |
| String sql = "SELECT b\"ByTes\""; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addByteArrayField("field1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues("ByTes".getBytes(StandardCharsets.UTF_8)).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSelectExcept() { |
| String sql = "SELECT * EXCEPT (Key, ts) FROM KeyValue;"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field2").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues("KeyValue234").build(), |
| Row.withSchema(schema).addValues("KeyValue235").build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSelectReplace() { |
| String sql = |
| "WITH orders AS\n" |
| + " (SELECT 5 as order_id,\n" |
| + " \"sprocket\" as item_name,\n" |
| + " 200 as quantity)\n" |
| + "SELECT * REPLACE (\"widget\" AS item_name)\n" |
| + "FROM orders"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addInt64Field("field1") |
| .addStringField("field2") |
| .addInt64Field("field3") |
| .build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues(5L, "widget", 200L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testUnionAllBasic() { |
| String sql = |
| "SELECT row_id FROM table_all_types UNION ALL SELECT row_id FROM table_all_types_2"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field1").build(); |
| |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValue(1L).build(), |
| Row.withSchema(schema).addValue(2L).build(), |
| Row.withSchema(schema).addValue(3L).build(), |
| Row.withSchema(schema).addValue(4L).build(), |
| Row.withSchema(schema).addValue(5L).build(), |
| Row.withSchema(schema).addValue(6L).build(), |
| Row.withSchema(schema).addValue(7L).build(), |
| Row.withSchema(schema).addValue(8L).build(), |
| Row.withSchema(schema).addValue(9L).build(), |
| Row.withSchema(schema).addValue(10L).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testAVGWithLongInput() { |
| String sql = "SELECT AVG(f_int_1) FROM aggregate_test_table;"; |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| thrown.expect(RuntimeException.class); |
| thrown.expectMessage( |
| "AVG(LONG) is not supported. You might want to use AVG(CAST(expression AS DOUBLE)."); |
| zetaSQLQueryPlanner.convertToBeamRel(sql); |
| } |
| |
| @Test |
| public void testReverseString() { |
| String sql = "SELECT REVERSE('abc');"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addStringField("field2").build(); |
| |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues("cba").build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCharLength() { |
| String sql = "SELECT CHAR_LENGTH('abc');"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addInt64Field("field").build(); |
| PAssert.that(stream).containsInAnyOrder(Row.withSchema(schema).addValues(3L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testCharLengthNull() { |
| String sql = "SELECT CHAR_LENGTH(@p0);"; |
| |
| ImmutableMap<String, Value> params = |
| ImmutableMap.of("p0", Value.createSimpleNullValue(TypeKind.TYPE_STRING)); |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = Schema.builder().addNullableField("field", FieldType.INT64).build(); |
| PAssert.that(stream) |
| .containsInAnyOrder(Row.withSchema(schema).addValues((Object) null).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testExtractTimestampThrowsOnMicrosecondNotSupported() { |
| String sql = |
| "WITH Timestamps AS (\n" |
| + " SELECT TIMESTAMP '2000-01-01 00:11:22.345678+00' as timestamp\n" |
| + ")\n" |
| + "SELECT\n" |
| + " timestamp,\n" |
| + " EXTRACT(ISOYEAR FROM timestamp) AS isoyear,\n" |
| + " EXTRACT(YEAR FROM timestamp) AS year,\n" |
| + " EXTRACT(ISOWEEK FROM timestamp) AS week,\n" |
| + " EXTRACT(MINUTE FROM timestamp) AS minute\n" |
| + "FROM Timestamps\n"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| ImmutableMap<String, Value> params = ImmutableMap.of(); |
| thrown.expect(IllegalArgumentException.class); |
| zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| } |
| |
| /** Only sample scenarios are covered here. Excessive testing is done via Compliance tests. */ |
| @Test |
| @Ignore("ZetaSQL does not support EnumType to IdentifierLiteral") |
| public void testExtractTimestamp() { |
| String sql = |
| "WITH Timestamps AS (\n" |
| + " SELECT TIMESTAMP '2005-01-03 12:34:56' AS timestamp UNION ALL\n" |
| + " SELECT TIMESTAMP '2017-05-26'\n" |
| + ")\n" |
| + "SELECT\n" |
| + " timestamp,\n" |
| + " EXTRACT(ISOYEAR FROM timestamp) AS isoyear,\n" |
| + " EXTRACT(YEAR FROM timestamp) AS year,\n" |
| + " EXTRACT(ISOWEEK FROM timestamp) AS week,\n" |
| + " EXTRACT(MINUTE FROM timestamp) AS minute\n" |
| + "FROM Timestamps\n"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| ImmutableMap<String, Value> params = ImmutableMap.of(); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addDateTimeField("ts") |
| .addField("isoyear", FieldType.INT64) |
| .addField("year", FieldType.INT64) |
| .addField("week", FieldType.INT64) |
| .addField("minute", FieldType.INT64) |
| .build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema) |
| .addValues( |
| DateTimeUtils.parseTimestampWithUTCTimeZone("2005-01-03 12:34:56"), |
| 2005L, |
| 2005L, |
| 1L, |
| 34L) |
| .build(), |
| Row.withSchema(schema) |
| .addValues(parseDate("2017-05-26"), 2017L, 2017L, 21L, 0L) |
| .build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| @Ignore("ZetaSQL does not support EnumType to IdentifierLiteral") |
| public void testExtractTimestampAtTimeZoneThrowsBecauseNotSupported() { |
| String sql = |
| "WITH Timestamps AS (\n" |
| + " SELECT TIMESTAMP '2017-05-26' AS timestamp\n" |
| + ")\n" |
| + "SELECT\n" |
| + " timestamp,\n" |
| + " EXTRACT(HOUR FROM timestamp AT TIME ZONE 'America/Vancouver') AS hour,\n" |
| + " EXTRACT(DAY FROM timestamp AT TIME ZONE 'America/Vancouver') AS day\n" |
| + "FROM Timestamps\n"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| ImmutableMap<String, Value> params = ImmutableMap.of(); |
| thrown.expect(IllegalArgumentException.class); |
| zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| } |
| |
| @Test |
| @Ignore("") |
| public void testExtractDateFromTimestampThrowsBecauseNotSupported() { |
| String sql = |
| "WITH Timestamps AS (\n" |
| + " SELECT TIMESTAMP '2017-05-26' AS ts\n" |
| + ")\n" |
| + "SELECT\n" |
| + " ts,\n" |
| + " EXTRACT(DATE FROM ts) AS dt\n" |
| + "FROM Timestamps\n"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| ImmutableMap<String, Value> params = ImmutableMap.of(); |
| thrown.expect(SqlException.class); |
| zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| } |
| |
| @Test |
| public void testIsNullTrueFalse() { |
| String sql = |
| "WITH Src AS (\n" |
| + " SELECT NULL as data UNION ALL\n" |
| + " SELECT TRUE UNION ALL\n" |
| + " SELECT FALSE\n" |
| + ")\n" |
| + "SELECT\n" |
| + " data IS NULL as isnull,\n" |
| + " data IS NOT NULL as isnotnull,\n" |
| + " data IS TRUE as istrue,\n" |
| + " data IS NOT TRUE as isnottrue,\n" |
| + " data IS FALSE as isfalse,\n" |
| + " data IS NOT FALSE as isnotfalse\n" |
| + "FROM Src\n"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| ImmutableMap<String, Value> params = ImmutableMap.of(); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql, params); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| final Schema schema = |
| Schema.builder() |
| .addField("isnull", FieldType.BOOLEAN) |
| .addField("isnotnull", FieldType.BOOLEAN) |
| .addField("istrue", FieldType.BOOLEAN) |
| .addField("isnottrue", FieldType.BOOLEAN) |
| .addField("isfalse", FieldType.BOOLEAN) |
| .addField("isnotfalse", FieldType.BOOLEAN) |
| .build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(schema).addValues(true, false, false, true, false, true).build(), |
| Row.withSchema(schema).addValues(false, true, true, false, false, true).build(), |
| Row.withSchema(schema).addValues(false, true, false, true, true, false).build()); |
| |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| @Test |
| public void testSimpleTableName() { |
| String sql = "SELECT Key FROM KeyValue"; |
| |
| ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config); |
| BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql); |
| PCollection<Row> stream = BeamSqlRelUtils.toPCollection(pipeline, beamRelNode); |
| |
| Schema singleField = Schema.builder().addInt64Field("field1").build(); |
| PAssert.that(stream) |
| .containsInAnyOrder( |
| Row.withSchema(singleField).addValues(14L).build(), |
| Row.withSchema(singleField).addValues(15L).build()); |
| pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES)); |
| } |
| |
| private void initializeCalciteEnvironment() { |
| initializeCalciteEnvironmentWithContext(); |
| } |
| |
| private void initializeCalciteEnvironmentWithContext(Context... extraContext) { |
| JdbcConnection jdbcConnection = |
| JdbcDriver.connect(tableProvider, PipelineOptionsFactory.create()); |
| SchemaPlus defaultSchemaPlus = jdbcConnection.getCurrentSchemaPlus(); |
| final ImmutableList<RelTraitDef> traitDefs = ImmutableList.of(ConventionTraitDef.INSTANCE); |
| |
| Object[] contexts = |
| ImmutableList.<Context>builder() |
| .add(Contexts.of(jdbcConnection.config())) |
| .add(extraContext) |
| .build() |
| .toArray(); |
| |
| this.config = |
| Frameworks.newConfigBuilder() |
| .defaultSchema(defaultSchemaPlus) |
| .traitDefs(traitDefs) |
| .context(Contexts.of(contexts)) |
| .ruleSets(BeamRuleSets.getRuleSets()) |
| .costFactory(null) |
| .typeSystem(jdbcConnection.getTypeFactory().getTypeSystem()) |
| .build(); |
| } |
| |
| private void initializeBeamTableProvider() { |
| Map<String, BeamSqlTable> testBoundedTableMap = new HashMap<>(); |
| testBoundedTableMap.put("KeyValue", BASIC_TABLE_ONE); |
| testBoundedTableMap.put("BigTable", BASIC_TABLE_TWO); |
| testBoundedTableMap.put("Spanner", BASIC_TABLE_THREE); |
| testBoundedTableMap.put("aggregate_test_table", AGGREGATE_TABLE_ONE); |
| testBoundedTableMap.put("window_test_table", TIMESTAMP_TABLE_ONE); |
| testBoundedTableMap.put("window_test_table_two", TIMESTAMP_TABLE_TWO); |
| testBoundedTableMap.put("time_test_table", TIME_TABLE); |
| testBoundedTableMap.put("all_null_table", TABLE_ALL_NULL); |
| testBoundedTableMap.put("table_with_struct", TABLE_WITH_STRUCT); |
| testBoundedTableMap.put("table_with_struct_two", TABLE_WITH_STRUCT_TWO); |
| testBoundedTableMap.put("table_with_array", TABLE_WITH_ARRAY); |
| testBoundedTableMap.put("table_with_array_for_unnest", TABLE_WITH_ARRAY_FOR_UNNEST); |
| testBoundedTableMap.put("table_for_case_when", TABLE_FOR_CASE_WHEN); |
| testBoundedTableMap.put("aggregate_test_table_two", AGGREGATE_TABLE_TWO); |
| testBoundedTableMap.put("table_empty", TABLE_EMPTY); |
| testBoundedTableMap.put("table_all_types", TABLE_ALL_TYPES); |
| testBoundedTableMap.put("table_all_types_2", TABLE_ALL_TYPES_2); |
| testBoundedTableMap.put("table_with_map", TABLE_WITH_MAP); |
| testBoundedTableMap.put("table_with_struct_ts_string", TABLE_WITH_STRUCT_TIMESTAMP_STRING); |
| |
| tableProvider = new ReadOnlyTableProvider("test_table_provider", testBoundedTableMap); |
| } |
| } |
