Merge pull request #1044 from data-integrations/PLUGIN-1257

[PLUGIN-1257]Fixed issue with deduplicate plugin failing when no filter condition is specified

Author: fernst

src/main/java/io/cdap/plugin/gcp/bigquery/sqlengine/builder/BigQueryBaseSQLBuilder.java

@@ -48,7 +48,9 @@ public abstract class BigQueryBaseSQLBuilder {
   public static final String ORDER_ASC = "ASC";
   public static final String SELECT_DEDUPLICATE_STATEMENT = "SELECT * EXCEPT(`%s`) FROM (%s) WHERE `%s` = 1";
   public static final String ROW_NUMBER_PARTITION_COLUMN =
-    "ROW_NUMBER() OVER ( PARTITION BY %s ORDER BY %s ) AS `%s`";
+    "ROW_NUMBER() OVER ( %s ) AS `%s`";
+  public static final String PARTITION_BY = "PARTITION BY ";
+  public static final String ORDER_BY = "ORDER BY ";
   public static final String NULLS_LAST = "NULLS LAST";
   public static final String IF_FUNCTION = "IF";
   public static final String ZERO = "0";

src/main/java/io/cdap/plugin/gcp/bigquery/sqlengine/builder/BigQueryDeduplicateSQLBuilder.java

@@ -96,11 +96,15 @@ protected String getSelectedFields(DeduplicateAggregationDefinition def) {
    */
   @VisibleForTesting
   protected String getRowNumColumn(DeduplicateAggregationDefinition def) {
-    String partitionByFields = getPartitionByFields(def.getGroupByExpressions());
-    String orderByFields = getOrderByFields(def.getFilterExpressions());
+    StringBuilder window = new StringBuilder();
+    // Add partition by clause for windowing
+    window.append(PARTITION_BY).append(getPartitionByFields(def.getGroupByExpressions()));
+    // Add ordering clause if specified
+    if (def.getFilterExpressions() != null && def.getFilterExpressions().size() > 0) {
+      window.append(SPACE).append(ORDER_BY).append(getOrderByFields(def.getFilterExpressions()));
+    }
     return String.format(ROW_NUMBER_PARTITION_COLUMN,
-                         partitionByFields,
-                         orderByFields,
+                         window.toString(),
                          rowNumColumnAlias);
   }
 

src/test/java/io/cdap/plugin/gcp/bigquery/sqlengine/builder/BigQueryDeduplicateSQLBuilderTest.java

@@ -38,12 +38,12 @@ public class BigQueryDeduplicateSQLBuilderTest {
   private Map<String, Expression> selectFields;
   private List<Expression> dedupFields;
   private List<DeduplicateAggregationDefinition.FilterExpression> filterFields;
-  private DeduplicateAggregationDefinition def;
+  private DeduplicateAggregationDefinition fullDefinition;
+  private DeduplicateAggregationDefinition onlyDedupFieldsDefinition;
 
   @Before
   public void setUp() {
     factory = new SQLExpressionFactory();
-    DeduplicateAggregationDefinition.Builder builder = DeduplicateAggregationDefinition.builder();
 
     // Build aggregation definition
     selectFields = new LinkedHashMap<>();
@@ -65,10 +65,17 @@ public void setUp() {
     filterFields.add(new DeduplicateAggregationDefinition.FilterExpression(
       factory.compile("f"), DeduplicateAggregationDefinition.FilterFunction.MIN));
 
-    builder.select(selectFields).dedupOn(dedupFields).filterDuplicatesBy(filterFields);
-    def = builder.build();
-
-    helper = new BigQueryDeduplicateSQLBuilder(def, "select * from tbl", "ds", "the_row_number");
+    fullDefinition = DeduplicateAggregationDefinition.builder()
+      .select(selectFields)
+      .dedupOn(dedupFields)
+      .filterDuplicatesBy(filterFields)
+      .build();
+    onlyDedupFieldsDefinition = DeduplicateAggregationDefinition.builder()
+      .select(selectFields)
+      .dedupOn(dedupFields)
+      .build();
+
+    helper = new BigQueryDeduplicateSQLBuilder(fullDefinition, "select * from tbl", "ds", "the_row_number");
   }
 
   @Test
@@ -114,13 +121,21 @@ public void testGetSelectedFields() {
         + "f AS f , "
         + "ROW_NUMBER() OVER ( PARTITION BY c , d , e ORDER BY e DESC NULLS LAST , f ASC NULLS LAST ) AS" +
         " `the_row_number`",
-      helper.getSelectedFields(def));
+      helper.getSelectedFields(fullDefinition));
   }
 
   @Test
   public void testGetRowNumColumn() {
     Assert.assertEquals("ROW_NUMBER() OVER ( PARTITION BY c , d , e ORDER BY e DESC NULLS LAST , " +
-                          "f ASC NULLS LAST ) AS `the_row_number`", helper.getRowNumColumn(def));
+                          "f ASC NULLS LAST ) AS `the_row_number`",
+                        helper.getRowNumColumn(fullDefinition));
+  }
+
+
+  @Test
+  public void testGetRowNumColumnWithoutOrderFields() {
+    Assert.assertEquals("ROW_NUMBER() OVER ( PARTITION BY c , d , e ) AS `the_row_number`",
+                        helper.getRowNumColumn(onlyDedupFieldsDefinition));
   }
 
   @Test