First Unbounded Source Implementation

flink-connector-bigquery-examples/pom.xml

@@ -41,6 +41,11 @@ under the License.
             <artifactId>flink-connector-bigquery</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-base</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-streaming-java</artifactId>

flink-connector-bigquery/src/main/java/com/google/cloud/flink/bigquery/services/BigQueryServices.java

@@ -124,6 +124,18 @@ interface QueryDataClient extends Serializable {
         Optional<TablePartitionInfo> retrievePartitionColumnInfo(
                 String project, String dataset, String table);
 
+        /**
+         * Returns, in case of being a partitioned table, all the partitions present alongside their
+         * status.
+         *
+         * @param project The GCP project.
+         * @param dataset The BigQuery dataset.
+         * @param table The BigQuery table.
+         * @return The information and status of the table's partitions.
+         */
+        List<PartitionIdWithInfoAndStatus> retrievePartitionsStatus(
+                String project, String dataset, String table);
+
         /**
          * Returns the {@link TableSchema} of the specified BigQuery table.
          *

flink-connector-bigquery/src/main/java/com/google/cloud/flink/bigquery/services/BigQueryServicesImpl.java

@@ -49,6 +49,8 @@
 import com.google.cloud.flink.bigquery.common.config.CredentialsOptions;
 import com.google.cloud.flink.bigquery.common.utils.BigQueryPartitionUtils;
 import com.google.cloud.flink.bigquery.common.utils.SchemaTransform;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.threeten.bp.Duration;
 
 import java.io.IOException;
@@ -64,6 +66,7 @@
 /** Implementation of the {@link BigQueryServices} interface that wraps the actual clients. */
 @Internal
 public class BigQueryServicesImpl implements BigQueryServices {
+    private static final Logger LOG = LoggerFactory.getLogger(BigQueryServicesImpl.class);
 
     @Override
     public StorageReadClient getStorageClient(CredentialsOptions credentialsOptions)
@@ -190,21 +193,24 @@ public List<String> retrieveTablePartitions(String project, String dataset, Stri
                                                 project, dataset),
                                         "WHERE",
                                         " partition_id <> '__STREAMING_UNPARTITIONED__'",
-                                        String.format(" table_catalog = '%s'", project),
+                                        String.format(" AND table_catalog = '%s'", project),
                                         String.format(" AND table_schema = '%s'", dataset),
                                         String.format(" AND table_name = '%s'", table),
-                                        "ORDER BY 1 DESC;")
+                                        "ORDER BY 1 ASC;")
                                 .stream()
                                 .collect(Collectors.joining("\n"));
 
                 QueryJobConfiguration queryConfig = QueryJobConfiguration.newBuilder(query).build();
 
                 TableResult results = bigQuery.query(queryConfig);
 
-                return StreamSupport.stream(results.iterateAll().spliterator(), false)
-                        .flatMap(row -> row.stream())
-                        .map(fValue -> fValue.getStringValue())
-                        .collect(Collectors.toList());
+                List<String> result =
+                        StreamSupport.stream(results.iterateAll().spliterator(), false)
+                                .flatMap(row -> row.stream())
+                                .map(fValue -> fValue.getStringValue())
+                                .collect(Collectors.toList());
+                LOG.info("Table partitions: {}", result);
+                return result;
             } catch (Exception ex) {
                 throw new RuntimeException(
                         String.format(
@@ -215,6 +221,7 @@ public List<String> retrieveTablePartitions(String project, String dataset, Stri
             }
         }
 
+        @Override
         public List<PartitionIdWithInfoAndStatus> retrievePartitionsStatus(
                 String project, String dataset, String table) {
             try {

flink-connector-bigquery/src/main/java/com/google/cloud/flink/bigquery/services/PartitionIdWithInfo.java

@@ -41,10 +41,7 @@ public TablePartitionInfo getInfo() {
 
     @Override
     public int hashCode() {
-        int hash = 5;
-        hash = 97 * hash + Objects.hashCode(this.getInfo());
-        hash = 97 * hash + Objects.hashCode(this.getPartitionId());
-        return hash;
+        return Objects.hash(this.partitionId, this.info);
     }
 
     @Override
@@ -59,10 +56,8 @@ public boolean equals(Object obj) {
             return false;
         }
         final PartitionIdWithInfo other = (PartitionIdWithInfo) obj;
-        if (!Objects.equals(this.getPartitionId(), other.getPartitionId())) {
-            return false;
-        }
-        return Objects.equals(this.getInfo(), other.getInfo());
+        return Objects.equals(this.getPartitionId(), other.getPartitionId())
+                && Objects.equals(this.getInfo(), other.getInfo());
     }
 
     @Override

flink-connector-bigquery/src/main/java/com/google/cloud/flink/bigquery/services/PartitionIdWithInfoAndStatus.java

@@ -50,13 +50,13 @@ public BigQueryPartitionUtils.PartitionStatus getStatus() {
         return status;
     }
 
+    public Boolean isCompleted() {
+        return status.equals(BigQueryPartitionUtils.PartitionStatus.COMPLETED);
+    }
+
     @Override
     public int hashCode() {
-        int hash = 7;
-        hash = 17 * hash + Objects.hashCode(this.getPartitionId());
-        hash = 17 * hash + Objects.hashCode(this.getInfo());
-        hash = 17 * hash + Objects.hashCode(this.getStatus());
-        return hash;
+        return Objects.hash(getPartitionId(), getInfo(), getStatus());
     }
 
     @Override
@@ -71,13 +71,9 @@ public boolean equals(Object obj) {
             return false;
         }
         final PartitionIdWithInfoAndStatus other = (PartitionIdWithInfoAndStatus) obj;
-        if (!Objects.equals(this.getPartitionId(), other.getPartitionId())) {
-            return false;
-        }
-        if (!Objects.equals(this.getInfo(), other.getInfo())) {
-            return false;
-        }
-        return this.getStatus() == other.getStatus();
+        return Objects.equals(this.getPartitionId(), other.getPartitionId())
+                && Objects.equals(this.getInfo(), other.getInfo())
+                && this.getStatus() == other.getStatus();
     }
 
     @Override

flink-connector-bigquery/src/main/java/com/google/cloud/flink/bigquery/services/TablePartitionInfo.java

@@ -25,6 +25,7 @@
 
 import java.time.Instant;
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
@@ -41,11 +42,11 @@ public TablePartitionInfo(
             String columnName,
             PartitionType partitionType,
             StandardSQLTypeName columnType,
-            Instant sbOldestEntryTime) {
+            Instant streamingBufferOldestEntryTime) {
         this.columnName = columnName;
         this.columnType = columnType;
         this.partitionType = partitionType;
-        this.streamingBufferOldestEntryTime = sbOldestEntryTime;
+        this.streamingBufferOldestEntryTime = streamingBufferOldestEntryTime;
     }
 
     public String getColumnName() {
@@ -74,6 +75,34 @@ public List<PartitionIdWithInfo> toPartitionsWithInfo(List<String> partitionIds)
                 .orElse(Lists.newArrayList());
     }
 
+    @Override
+    public int hashCode() {
+        return Objects.hash(
+                this.columnName,
+                this.columnType,
+                this.partitionType,
+                this.streamingBufferOldestEntryTime);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj == null) {
+            return false;
+        }
+        if (getClass() != obj.getClass()) {
+            return false;
+        }
+        final TablePartitionInfo other = (TablePartitionInfo) obj;
+        return Objects.equals(this.columnName, other.columnName)
+                && this.columnType == other.columnType
+                && this.partitionType == other.partitionType
+                && Objects.equals(
+                        this.streamingBufferOldestEntryTime, other.streamingBufferOldestEntryTime);
+    }
+
     @Override
     public String toString() {
         return "TablePartitionInfo{"
@@ -83,6 +112,8 @@ public String toString() {
                 + columnType
                 + ", partitionType="
                 + partitionType
+                + ", streamingBufferOldestEntryTime="
+                + streamingBufferOldestEntryTime
                 + '}';
     }
 }