Better formatting/logging for pre-sync data (#25459)

* Log cursor alongside indexes

* Better formatting for table estimates, indexes

* Doc & dockerfile bump

* Fix formatting

* auto-bump connector version

* Automated Commit - Formatting Changes

* Fixing connector base issues

* Fix failure

* Fix changelog

---------

Co-authored-by: Octavia Squidington III <[email protected]>
Co-authored-by: airbyteio <[email protected]>

Author: 3 people

airbyte-config-oss/init-oss/src/main/resources/seed/oss_catalog.json

@@ -21220,7 +21220,7 @@
     "sourceDefinitionId": "decd338e-5647-4c0b-adf4-da0e75f5a750",
     "name": "Postgres",
     "dockerRepository": "airbyte/source-postgres",
-    "dockerImageTag": "2.0.24",
+    "dockerImageTag": "2.0.25",
     "documentationUrl": "https://docs.airbyte.com/integrations/sources/postgres",
     "icon": "postgresql.svg",
     "sourceType": "database",

airbyte-config-oss/init-oss/src/main/resources/seed/source_definitions.yaml

@@ -1619,7 +1619,7 @@
 - name: Postgres
   sourceDefinitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
   dockerRepository: airbyte/source-postgres
-  dockerImageTag: 2.0.24
+  dockerImageTag: 2.0.25
   documentationUrl: https://docs.airbyte.com/integrations/sources/postgres
   icon: postgresql.svg
   sourceType: database

airbyte-config-oss/init-oss/src/main/resources/seed/source_specs.yaml

@@ -12152,7 +12152,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-postgres:2.0.24"
+- dockerImage: "airbyte/source-postgres:2.0.25"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/sources/postgres"
     connectionSpecification:

airbyte-db/db-lib/src/main/java/io/airbyte/db/jdbc/JdbcUtils.java

@@ -25,6 +25,8 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.Maps;
 import java.sql.JDBCType;
+import java.text.CharacterIterator;
+import java.text.StringCharacterIterator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -135,4 +137,41 @@ public static boolean useSsl(final JsonNode config) {
       return config.get(SSL_KEY).asBoolean();
   }
 
+  /**
+   * Helper method for logging bytes in a human-readable format. This method logs in SI units
+   * (B/kB/MB/GB)
+   */
+  public static String humanReadableByteCountSI(final long bytes) {
+    if (-1000 < bytes && bytes < 1000) {
+      return bytes + " B";
+    }
+    final CharacterIterator ci = new StringCharacterIterator("kMGTPE");
+    long formattedBytes = bytes;
+    while (bytes <= -999_950 || bytes >= 999_950) {
+      formattedBytes /= 1000;
+      ci.next();
+    }
+    return String.format("%.1f %cB", formattedBytes / 1000.0, ci.current());
+  }
+
+  /**
+   * Helper method for logging bytes in a human-readable format. This method logs in Binary units
+   * (B/KiB/MiB/GiB)
+   */
+  public static String humanReadableByteCountBin(final long bytes) {
+    final long absB = bytes == Long.MIN_VALUE ? Long.MAX_VALUE : Math.abs(bytes);
+    final int bytePrefixCount = 1024;
+    if (absB < bytePrefixCount) {
+      return bytes + " B";
+    }
+    long value = absB;
+    final CharacterIterator ci = new StringCharacterIterator("KMGTPE");
+    for (int i = 40; i >= 0 && absB > 0xfffccccccccccccL >> i; i -= 10) {
+      value >>= 10;
+      ci.next();
+    }
+    value *= Long.signum(bytes);
+    return String.format("%.1f %ciB", value / 1024.0, ci.current());
+  }
+
 }

airbyte-integrations/connectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java

@@ -442,21 +442,24 @@ protected void logPreSyncDebugData(final JdbcDatabase database, final Configured
         database.getMetaData().getDatabaseProductVersion());
 
     for (final ConfiguredAirbyteStream stream : catalog.getStreams()) {
-      final String streamName = stream.getStream().getName();
-      final String schemaName = stream.getStream().getNamespace();
-      final ResultSet indexInfo = database.getMetaData().getIndexInfo(null,
-          schemaName,
-          streamName,
-          false,
-          false);
-      LOGGER.info("Discovering indexes for schema \"{}\", table \"{}\"", schemaName, streamName);
-      while (indexInfo.next()) {
-        LOGGER.info("Index name: {}, Column: {}, Unique: {}",
-            indexInfo.getString(JDBC_INDEX_NAME),
-            indexInfo.getString(JDBC_COLUMN_COLUMN_NAME),
-            !indexInfo.getBoolean(JDBC_INDEX_NON_UNIQUE));
+      if (stream.getSyncMode().equals(SyncMode.INCREMENTAL)) {
+        final String streamName = stream.getStream().getName();
+        final String schemaName = stream.getStream().getNamespace();
+        final String cursorFieldName = stream.getCursorField() != null && stream.getCursorField().size() != 0 ? stream.getCursorField().get(0) : "";
+        final ResultSet indexInfo = database.getMetaData().getIndexInfo(null,
+            schemaName,
+            streamName,
+            false,
+            false);
+        LOGGER.info("Discovering indexes for schema \"{}\", table \"{}\", with cursor field \"{}\"", schemaName, streamName, cursorFieldName);
+        while (indexInfo.next()) {
+          LOGGER.info("Index name: {}, Column: {}, Unique: {}",
+              indexInfo.getString(JDBC_INDEX_NAME),
+              indexInfo.getString(JDBC_COLUMN_COLUMN_NAME),
+              !indexInfo.getBoolean(JDBC_INDEX_NON_UNIQUE));
+        }
+        indexInfo.close();
       }
-      indexInfo.close();
     }
   }
 

airbyte-integrations/connectors/source-postgres-strict-encrypt/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-postgres-strict-encrypt
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=2.0.24
+LABEL io.airbyte.version=2.0.25
 LABEL io.airbyte.name=airbyte/source-postgres-strict-encrypt

airbyte-integrations/connectors/source-postgres/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-postgres
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=2.0.24
+LABEL io.airbyte.version=2.0.25
 LABEL io.airbyte.name=airbyte/source-postgres

airbyte-integrations/connectors/source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSource.java

@@ -74,6 +74,7 @@
 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.text.NumberFormat;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
@@ -572,8 +573,8 @@ protected void estimateFullRefreshSyncSize(final JdbcDatabase database,
         // read a row and Stringify it to better understand the accurate volume of data sent over the wire.
         // However, this approach doesn't account for different row sizes.
         AirbyteTraceMessageUtility.emitEstimateTrace(PLATFORM_DATA_INCREASE_FACTOR * syncByteCount, Type.STREAM, syncRowCount, tableName, schemaName);
-        LOGGER.info(String.format("Estimate for table: %s : {sync_row_count: %s, sync_bytes: %s, total_table_row_count: %s, total_table_bytes: %s}",
-            fullTableName, syncRowCount, syncByteCount, syncRowCount, syncByteCount));
+        LOGGER.info(String.format("Estimate for table in full refresh mode: %s : {rows_to_sync: %s, data_to_sync: %s}",
+            fullTableName, NumberFormat.getInstance().format(syncRowCount), JdbcUtils.humanReadableByteCountSI(syncByteCount)));
       }
     } catch (final SQLException e) {
       LOGGER.warn("Error occurred while attempting to estimate sync size", e);
@@ -614,8 +615,9 @@ protected void estimateIncrementalSyncSize(final JdbcDatabase database,
       // read a row and Stringify it to better understand the accurate volume of data sent over the wire.
       // However, this approach doesn't account for different row sizes
       AirbyteTraceMessageUtility.emitEstimateTrace(PLATFORM_DATA_INCREASE_FACTOR * syncByteCount, Type.STREAM, syncRowCount, tableName, schemaName);
-      LOGGER.info(String.format("Estimate for table: %s : {sync_row_count: %s, sync_bytes: %s, total_table_row_count: %s, total_table_bytes: %s}",
-          fullTableName, syncRowCount, syncByteCount, tableRowCount, tableRowCount));
+      LOGGER.info(String.format("Estimate for table in incremental mode: %s : {rows_to_sync: %s, data_to_sync: %s, table_row_count: %s, table_size: %s}",
+          fullTableName, NumberFormat.getInstance().format(syncRowCount), JdbcUtils.humanReadableByteCountSI(syncByteCount), NumberFormat.getInstance().format(tableRowCount),
+          JdbcUtils.humanReadableByteCountSI(tableByteCount)));
     } catch (final SQLException e) {
       LOGGER.warn("Error occurred while attempting to estimate sync size", e);
     }

connectors.md

@@ -183,7 +183,7 @@
 | **PokeAPI** | <img alt="PokeAPI icon" src="https://raw.githubusercontent.com/airbytehq/airbyte /master/airbyte-config-oss/init-oss/src/main/resources/icons/pokeapi.svg" height="30" height="30"/> | Source | airbyte/source-pokeapi:0.1.5 | alpha | [docs](https://docs.airbyte.com/integrations/sources/pokeapi) | [connectors/source/pokeapi](https://github.com/airbytehq/airbyte/issues?q=is:open+is:issue+label:connectors/source/pokeapi) | [source-pokeapi](https://github.com/airbytehq/airbyte/tree/master/airbyte-integrations/connectors/source-pokeapi) | <small>`6371b14b-bc68-4236-bfbd-468e8df8e968`</small> |
 | **Polygon Stock API** | <img alt="Polygon Stock API icon" src="https://raw.githubusercontent.com/airbytehq/airbyte /master/airbyte-config-oss/init-oss/src/main/resources/icons/polygon.svg" height="30" height="30"/> | Source | airbyte/source-polygon-stock-api:0.1.1 | alpha | [docs](https://docs.airbyte.com/integrations/sources/polygon-stock-api) | [connectors/source/polygon-stock-api](https://github.com/airbytehq/airbyte/issues?q=is:open+is:issue+label:connectors/source/polygon-stock-api) | [source-polygon-stock-api](https://github.com/airbytehq/airbyte/tree/master/airbyte-integrations/connectors/source-polygon-stock-api) | <small>`5807d72f-0abc-49f9-8fa5-ae820007032b`</small> |
 | **PostHog** | <img alt="PostHog icon" src="https://raw.githubusercontent.com/airbytehq/airbyte /master/airbyte-config-oss/init-oss/src/main/resources/icons/posthog.svg" height="30" height="30"/> | Source | airbyte/source-posthog:0.1.9 | beta | [docs](https://docs.airbyte.com/integrations/sources/posthog) | [connectors/source/posthog](https://github.com/airbytehq/airbyte/issues?q=is:open+is:issue+label:connectors/source/posthog) | [source-posthog](https://github.com/airbytehq/airbyte/tree/master/airbyte-integrations/connectors/source-posthog) | <small>`af6d50ee-dddf-4126-a8ee-7faee990774f`</small> |
-| **Postgres** | <img alt="Postgres icon" src="https://raw.githubusercontent.com/airbytehq/airbyte /master/airbyte-config-oss/init-oss/src/main/resources/icons/postgresql.svg" height="30" height="30"/> | Source | airbyte/source-postgres:2.0.24 | generally_available | [docs](https://docs.airbyte.com/integrations/sources/postgres) | [connectors/source/postgres](https://github.com/airbytehq/airbyte/issues?q=is:open+is:issue+label:connectors/source/postgres) | [source-postgres](https://github.com/airbytehq/airbyte/tree/master/airbyte-integrations/connectors/source-postgres) | <small>`decd338e-5647-4c0b-adf4-da0e75f5a750`</small> |
+| **Postgres** | <img alt="Postgres icon" src="https://raw.githubusercontent.com/airbytehq/airbyte /master/airbyte-config-oss/init-oss/src/main/resources/icons/postgresql.svg" height="30" height="30"/> | Source | airbyte/source-postgres:2.0.25 | generally_available | [docs](https://docs.airbyte.com/integrations/sources/postgres) | [connectors/source/postgres](https://github.com/airbytehq/airbyte/issues?q=is:open+is:issue+label:connectors/source/postgres) | [source-postgres](https://github.com/airbytehq/airbyte/tree/master/airbyte-integrations/connectors/source-postgres) | <small>`decd338e-5647-4c0b-adf4-da0e75f5a750`</small> |
 | **Postmark App** | <img alt="Postmark App icon" src="https://raw.githubusercontent.com/airbytehq/airbyte /master/airbyte-config-oss/init-oss/src/main/resources/icons/postmark.svg" height="30" height="30"/> | Source | airbyte/source-postmarkapp:0.1.0 | alpha | [docs](https://docs.airbyte.com/integrations/sources/postmarkapp) | [connectors/source/postmarkapp](https://github.com/airbytehq/airbyte/issues?q=is:open+is:issue+label:connectors/source/postmarkapp) | [source-postmarkapp](https://github.com/airbytehq/airbyte/tree/master/airbyte-integrations/connectors/source-postmarkapp) | <small>`cde75ca1-1e28-4a0f-85bb-90c546de9f1f`</small> |
 | **PrestaShop** | <img alt="PrestaShop icon" src="https://raw.githubusercontent.com/airbytehq/airbyte /master/airbyte-config-oss/init-oss/src/main/resources/icons/prestashop.svg" height="30" height="30"/> | Source | airbyte/source-prestashop:0.3.1 | beta | [docs](https://docs.airbyte.com/integrations/sources/prestashop) | [connectors/source/prestashop](https://github.com/airbytehq/airbyte/issues?q=is:open+is:issue+label:connectors/source/prestashop) | [source-prestashop](https://github.com/airbytehq/airbyte/tree/master/airbyte-integrations/connectors/source-prestashop) | <small>`d60a46d4-709f-4092-a6b7-2457f7d455f5`</small> |
 | **Primetric** | <img alt="Primetric icon" src="https://raw.githubusercontent.com/airbytehq/airbyte /master/airbyte-config-oss/init-oss/src/main/resources/icons/primetric.svg" height="30" height="30"/> | Source | airbyte/source-primetric:0.1.0 | alpha | [docs](https://docs.airbyte.com/integrations/sources/primetric) | [connectors/source/primetric](https://github.com/airbytehq/airbyte/issues?q=is:open+is:issue+label:connectors/source/primetric) | [source-primetric](https://github.com/airbytehq/airbyte/tree/master/airbyte-integrations/connectors/source-primetric) | <small>`f636c3c6-4077-45ac-b109-19fc62a283c1`</small> |