Merge pull request #1399 from cloudsufi/cherry-pick-05b4423040c03237da1afa4c54d31f5aef60b505

[🍒][PLUGIN-788] Fixed Partition Field Validation

Author: itsankit-google

src/main/java/io/cdap/plugin/gcp/bigquery/sink/BigQuerySinkConfig.java

@@ -390,6 +390,10 @@ private void validatePartitionProperties(@Nullable Schema schema, FailureCollect
     String tableName = getTable();
     String serviceAccount = getServiceAccount();
 
+    if (shouldCreatePartitionedTable()) {
+      validateColumnForPartition(partitionByField, schema, collector);
+    }
+
     if (project == null || dataset == null || tableName == null || serviceAccount == null) {
       return;
     }
@@ -415,11 +419,6 @@ private void validatePartitionProperties(@Nullable Schema schema, FailureCollect
       } else if (rangePartitioning != null) {
         validateRangePartitionTableWithInputConfiguration(table, rangePartitioning, collector);
       }
-      validateColumnForPartition(partitionByField, schema, collector);
-      return;
-    }
-    if (shouldCreatePartitionedTable()) {
-      validateColumnForPartition(partitionByField, schema, collector);
     }
   }
 
@@ -540,6 +539,14 @@ private void validateIntegerPartitioningRange(Long rangeStart, Long rangeEnd, Lo
         .withConfigProperty(NAME_RANGE_END);
     }
 
+    if (!containsMacro(NAME_RANGE_START) && !containsMacro(NAME_RANGE_END) && rangeStart != null && rangeEnd != null) {
+      if (rangeEnd <= rangeStart) {
+        collector.addFailure("Range End is less than or equal to Range Start.",
+                             "Range End must be greater than Range Start.")
+          .withConfigProperty(NAME_RANGE_END);
+      }
+    }
+
     if (!containsMacro(NAME_RANGE_INTERVAL)) {
       if (rangeInterval == null) {
         collector.addFailure(