/*
 * 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.example;

import java.sql.Types;
import java.util.Arrays;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.extensions.sql.BeamSql;
import org.apache.beam.sdk.extensions.sql.schema.BeamRecordSqlType;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.BeamRecord;
import org.apache.beam.sdk.values.PBegin;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTag;

/**
 * This is a quick example, which uses Beam SQL DSL to create a data pipeline.
 *
 * <p>Run the example with
 * <pre>
 * mvn -pl dsls/sql compile exec:java \
 *  -Dexec.mainClass=BeamSqlExample \
 *   -Dexec.args="--runner=DirectRunner" -Pdirect-runner
 * </pre>
 *
 */
class BeamSqlExample {
  public static void main(String[] args) throws Exception {
    PipelineOptions options = PipelineOptionsFactory.fromArgs(args).as(PipelineOptions.class);
    Pipeline p = Pipeline.create(options);

    //define the input row format
    List<String> fieldNames = Arrays.asList("c1", "c2", "c3");
    List<Integer> fieldTypes = Arrays.asList(Types.INTEGER, Types.VARCHAR, Types.DOUBLE);
    BeamRecordSqlType type = BeamRecordSqlType.create(fieldNames, fieldTypes);
    BeamRecord row = new BeamRecord(type, 1, "row", 1.0);

    //create a source PCollection with Create.of();
    PCollection<BeamRecord> inputTable = PBegin.in(p).apply(Create.of(row)
        .withCoder(type.getRecordCoder()));

    //Case 1. run a simple SQL query over input PCollection with BeamSql.simpleQuery;
    PCollection<BeamRecord> outputStream = inputTable.apply(
        BeamSql.simpleQuery("select c1, c2, c3 from PCOLLECTION where c1=1"));

    //print the output record of case 1;
    outputStream.apply("log_result",
        MapElements.<BeamRecord, Void>via(new SimpleFunction<BeamRecord, Void>() {
      public Void apply(BeamRecord input) {
        System.out.println("PCOLLECTION: " + input);
        return null;
      }
    }));

    //Case 2. run the query with BeamSql.query over result PCollection of case 1.
    PCollection<BeamRecord> outputStream2 =
        PCollectionTuple.of(new TupleTag<BeamRecord>("CASE1_RESULT"), outputStream)
        .apply(BeamSql.query("select c2, c3 from CASE1_RESULT where c1=1"));

    //print the output record of case 2;
    outputStream2.apply("log_result",
        MapElements.<BeamRecord, Void>via(new SimpleFunction<BeamRecord, Void>() {
      @Override
      public Void apply(BeamRecord input) {
        System.out.println("TABLE_B: " + input);
        return null;
      }
    }));

    p.run().waitUntilFinish();
  }
}

@@ -1710,16 +1710,39 @@
 -pl 
-dsls/sql
+sdks/java/extensions/sql %5C%0A *  
  com
@@ -1756,22 +1756,16 @@
 ec:java 
-%5C%0A *  
 -Dexec.m
@@ -1773,16 +1773,59 @@
 inClass=
+org.apache.beam.sdk.extensions.sql.example.
 BeamSqlE
@@ -1900,19 +1900,16 @@
  %3C/pre%3E%0A
- *%0A
  */%0Aclas
@@ -2410,16 +2410,17 @@
 cord row
+1
  = new B
@@ -2450,16 +2450,134 @@
 %22, 1.0);
+%0A    BeamRecord row2 = new BeamRecord(type, 2, %22row%22, 2.0);%0A    BeamRecord row3 = new BeamRecord(type, 3, %22row%22, 3.0);
 %0A%0A    //
@@ -2695,16 +2695,29 @@
 e.of(row
+1, row2, row3
 )%0A      
@@ -2966,33 +2966,35 @@
 LECTION where c1
-=
+ %3E 
 1%22));%0A%0A    //pri
@@ -3181,32 +3181,135 @@
 Record input) %7B%0A
+        //expect output:%0A        //  PCOLLECTION: %5B3, row, 3.0%5D%0A        //  PCOLLECTION: %5B2, row, 2.0%5D%0A
         System.o
@@ -3334,32 +3334,48 @@
 CTION: %22 + input
+.getDataValues()
 );%0A        retur
@@ -3647,26 +3647,31 @@
 %22select c2, 
+sum(
 c3
+)
  from CASE1_
@@ -3681,18 +3681,19 @@
 ULT 
-where c1=1
+group by c2
 %22));
@@ -3928,46 +3928,129 @@
     
-System.out.println(%22TABLE_B: %22 + input
+//expect output:%0A        //  CASE1_RESULT: %5Brow, 5.0%5D%0A        System.out.println(%22CASE1_RESULT: %22 + input.getDataValues()
 );%0A 

Update BeamSqlExample: - Fix mvn example - Add aggregation/group by