oceanprotocol · idiom-bytes · Jul 31, 2024 · May 29, 2024 · May 30, 2024 · Jun 5, 2024
diff --git a/READMEs/diagrams/lake.html b/READMEs/diagrams/lake.html
@@ -0,0 +1,76 @@
+<html>
+	<head>
+		<title>ETL Architecture</title>
+		<style>
+			body {
+				font-family: Arial, sans-serif;
+				text-align: center;
+				background-color: #f0f0f0;
+				margin: 40px;
+			}
+
+			.banner {
+				font-size: 24px;
+				margin-bottom: 20px;
+			}
+
+			.pink {
+				color: rgb(255, 64, 146);
+			}
+		</style>
+		<script src="https://cdn.jsdelivr.net/npm/mermaid@8/dist/mermaid.min.js"></script>
+		<script>
+			mermaid.initialize({
+				startOnLoad: true,
+				theme: 'forest',
+			});
+		</script>
+	</head>
+	<body>
+		<div class="banner">
+			Predict<span class="pink">oo</span>r Tables & Lake
+		</div>
+		<div class="mermaid">
+			erDiagram
+				SUBGRAPH_predictPredictions ||--o{ PDR_PREDICTIONS : yields
+				SUBGRAPH_predictPayouts ||--o{ PDR_PAYOUTS : yields
+                PDR_PREDICTIONS ||--o{ BRONZE_PDR_PREDICTIONS : yields
+                PDR_PAYOUTS ||--o{ BRONZE_PDR_PREDICTIONS : yields
+				PDR_PREDICTIONS {
+					string ID
+                    string contract
+					int slot
+					string user
+                    string pair
+                    string timeframe
+                    string source
+					int timestamp
+				}
+				PDR_PAYOUTS {
+					string ID
+                    string contract
+                    int slot
+					string user
+                    boolean predvalue
+                    float stake
+                    float payout
+                    int timestamp
+				}
+				BRONZE_PDR_PREDICTIONS {
+                    string ID
+                    string slot_id
+                    string contract
+                    int slot
+                    string user
+                    string pair
+                    string timeframe
+                    string source
+                    boolean predvalue
+                    float stake
+                    float payout
+                    int timestamp
+                    int last_event_timestamp
+				}
+		</div>
+	</body>
+</html>
diff --git a/READMEs/images/etl_checkpoints.png b/READMEs/images/etl_checkpoints.png
diff --git a/READMEs/images/etl_swap_from_temp_to_prod_atomic.png b/READMEs/images/etl_swap_from_temp_to_prod_atomic.png
diff --git a/READMEs/images/gql_data_factory_fetch.png b/READMEs/images/gql_data_factory_fetch.png
diff --git a/READMEs/images/gql_data_factory_fetch_1_write_n.png b/READMEs/images/gql_data_factory_fetch_1_write_n.png
diff --git a/READMEs/images/gql_use_cache_and_swap.png b/READMEs/images/gql_use_cache_and_swap.png
diff --git a/READMEs/images/lake_tables_diagram.png b/READMEs/images/lake_tables_diagram.png
diff --git a/READMEs/lake-and-etl.md b/READMEs/lake-and-etl.md
diff --git a/pdr_backend/analytics/get_predictions_info.py b/pdr_backend/analytics/get_predictions_info.py
@@ -17,7 +17,7 @@
     plot_traction_daily_statistics,
 )
 from pdr_backend.lake.duckdb_data_store import DuckDBDataStore
-from pdr_backend.lake.table import NamedTable
+from pdr_backend.lake.table import Table
 from pdr_backend.ppss.ppss import PPSS
 from pdr_backend.util.time_types import UnixTimeMs
 
@@ -45,7 +45,7 @@ def get_predictions_info_main(
     logger.info("get_predictions_info_main start_timestr %s", start_timestr)
     logger.info("get_predictions_info_main end_timestr %s", end_timestr)
 
-    table_name = NamedTable("pdr_predictions").fullname
+    table_name = Table("pdr_predictions").table_name
 
     # convert feed addresses to string for SQL query
     feed_addrs_str = _address_list_to_str(feed_addrs)
@@ -74,7 +74,7 @@ def get_predictoors_info_main(
     logger.info(
         "get_predictoors_info_main_ppss.lake_ss.lake_dir--- %s", ppss.lake_ss.lake_dir
     )
-    table_name = NamedTable("pdr_predictions").fullname
+    table_name = Table("pdr_predictions").table_name
 
     # convert feed addresses to string for SQL query
     pdr_addrs_str = _address_list_to_str(pdr_addrs)
@@ -98,7 +98,7 @@ def get_predictoors_info_main(
 
 @enforce_types
 def get_traction_info_main(ppss: PPSS, start_timestr: str, end_timestr: str):
-    table_name = NamedTable("pdr_predictions").fullname
+    table_name = Table("pdr_predictions").table_name
 
     query = f"""
         SELECT *,

diff --git a/pdr_backend/analytics/test/test_get_predictions_info.py b/pdr_backend/analytics/test/test_get_predictions_info.py
@@ -9,7 +9,7 @@
 from enforce_typing import enforce_types
 from pdr_backend.analytics.get_predictions_info import get_predictions_info_main
 from pdr_backend.lake.prediction import Prediction
-from pdr_backend.lake.table import NamedTable
+from pdr_backend.lake.table import Table
 from pdr_backend.ppss.ppss import mock_ppss
 
 
@@ -34,7 +34,7 @@ def test_get_predictions_info_main_mainnet(
         fin_timestr=fin_timestr,
     )
     predictions_df = _gql_datafactory_first_predictions_df
-    predictions_table = NamedTable.from_dataclass(Prediction)
+    predictions_table = Table.from_dataclass(Prediction)
     predictions_table.append_to_storage(predictions_df, ppss)
 
     feed_addr = "0x2d8e2267779d27c2b3ed5408408ff15d9f3a3152"
@@ -99,7 +99,7 @@ def test_get_predictions_info_bad_date_range(
     )
 
     predictions_df = _gql_datafactory_first_predictions_df
-    predictions_table = NamedTable.from_dataclass(Prediction)
+    predictions_table = Table.from_dataclass(Prediction)
     predictions_table.append_to_storage(predictions_df, ppss)
 
     feed_addr = "0x2d8e2267779d27c2b3ed5408408ff15d9f3a3152"
@@ -154,7 +154,7 @@ def test_get_predictions_info_bad_feed(
     )
 
     predictions_df = _gql_datafactory_first_predictions_df
-    predictions_table = NamedTable.from_dataclass(Prediction)
+    predictions_table = Table.from_dataclass(Prediction)
     predictions_table.append_to_storage(predictions_df, ppss)
 
     feed_addr = "0x8e0we267779d27c2b3ed5408408ff15d9f3a3152"
@@ -196,7 +196,7 @@ def test_get_predictions_info_empty(_gql_datafactory_first_predictions_df, tmpdi
         fin_timestr=fin_timestr,
     )
 
-    predictions_table = NamedTable.from_dataclass(Prediction)
+    predictions_table = Table.from_dataclass(Prediction)
     predictions_table.append_to_storage(
         pl.DataFrame([], schema=Prediction.get_lake_schema()), ppss
     )

diff --git a/pdr_backend/analytics/test/test_get_predictoors_info.py b/pdr_backend/analytics/test/test_get_predictoors_info.py
@@ -10,7 +10,7 @@
 
 from pdr_backend.analytics.get_predictions_info import get_predictoors_info_main
 from pdr_backend.lake.prediction import Prediction
-from pdr_backend.lake.table import NamedTable
+from pdr_backend.lake.table import Table
 from pdr_backend.ppss.ppss import mock_ppss
 
 
@@ -32,7 +32,7 @@ def test_get_predictoors_info_main_mainnet(
     )
 
     predictions_df = _gql_datafactory_first_predictions_df
-    predictions_table = NamedTable.from_dataclass(Prediction)
+    predictions_table = Table.from_dataclass(Prediction)
     predictions_table.append_to_storage(predictions_df, ppss)
 
     user_addr = "0xaaaa4cb4ff2584bad80ff5f109034a891c3d88dd"
@@ -84,7 +84,7 @@ def test_get_predictoors_info_bad_date_range(
     )
 
     predictions_df = _gql_datafactory_first_predictions_df
-    predictions_table = NamedTable.from_dataclass(Prediction)
+    predictions_table = Table.from_dataclass(Prediction)
     predictions_table.append_to_storage(predictions_df, ppss)
 
     user_addr = "0xaaaa4cb4ff2584bad80ff5f109034a891c3d88dd"
@@ -134,7 +134,7 @@ def test_get_predictoors_info_bad_user_address(
     )
 
     predictions_df = _gql_datafactory_first_predictions_df
-    predictions_table = NamedTable.from_dataclass(Prediction)
+    predictions_table = Table.from_dataclass(Prediction)
     predictions_table.append_to_storage(predictions_df, ppss)
 
     user_addr = "0xbbbb4cb4ff2584bad80ff5f109034a891c3d223"

diff --git a/pdr_backend/analytics/test/test_get_traction_info.py b/pdr_backend/analytics/test/test_get_traction_info.py
@@ -10,7 +10,7 @@
 
 from pdr_backend.analytics.get_predictions_info import get_traction_info_main
 from pdr_backend.lake.prediction import Prediction
-from pdr_backend.lake.table import NamedTable
+from pdr_backend.lake.table import Table
 from pdr_backend.ppss.ppss import mock_ppss
 
 
@@ -40,7 +40,7 @@ def test_get_traction_info_main_mainnet(
     )
 
     predictions_df = _gql_datafactory_daily_predictions_df
-    predictions_table = NamedTable.from_dataclass(Prediction)
+    predictions_table = Table.from_dataclass(Prediction)
     predictions_table.append_to_storage(predictions_df, ppss)
 
     get_traction_info_main(ppss, st_timestr, fin_timestr)
@@ -80,7 +80,7 @@ def test_get_traction_info_empty_data(
         fin_timestr=fin_timestr,
     )
 
-    pdr_prediction_table = NamedTable.from_dataclass(Prediction)
+    pdr_prediction_table = Table.from_dataclass(Prediction)
     pdr_prediction_table.append_to_storage(
         pl.DataFrame([], schema=Prediction.get_lake_schema()), ppss
     )

diff --git a/pdr_backend/cli/cli_arguments.py b/pdr_backend/cli/cli_arguments.py
@@ -48,9 +48,7 @@
   pdr arima_plots PPSS_FILE [--debug_mode False]
   pdr deployer (for >1 predictoor bots)
   pdr lake raw|etl update PPSS_FILE NETWORK
-  pdr lake raw|etl drop PPSS_FILE NETWORK ST
   pdr lake describe --HTML PPSS_FILE NETWORK
-  pdr lake validate PPSS_FILE NETWORK
   pdr analytics PPSS_FILE NETWORK
 
 Utilities:

diff --git a/pdr_backend/cli/cli_module_lake.py b/pdr_backend/cli/cli_module_lake.py
@@ -71,6 +71,7 @@ def do_lake_query(args, ppss):
     try:
         df = db.query_data(args.QUERY)
         print(df)
+        print("Rows:", len(df))
     except Exception as e:
         logger.error("Error querying lake: %s", e)
         print(e)

diff --git a/pdr_backend/cli/test/test_cli_module_lake.py b/pdr_backend/cli/test/test_cli_module_lake.py
@@ -173,11 +173,11 @@ def test_do_lake_raw_drop(tmpdir, caplog):
 
     assert "drop table _temp_test1 starting at 1609459200000" in caplog.text
     assert "rows before: 5" in caplog.text
-    assert "rows after: 2" in caplog.text
+    assert "rows after: 3" in caplog.text
     assert "drop table test2 starting at 1609459200000" in caplog.text
     assert "rows before: 5" in caplog.text
     assert "rows after: 3" in caplog.text
-    assert "truncated 5 rows from 2 tables" in caplog.text
+    assert "truncated 8 rows from 3 tables" in caplog.text
 
 
 @enforce_types
@@ -193,8 +193,8 @@ def test_do_lake_etl_drop(tmpdir, caplog):
 
     db = DuckDBDataStore(str(tmpdir))
     _make_and_fill_timestamps(db, "_temp_bronze_test1", ts - 3 * one_day)
-    _make_and_fill_timestamps(db, "_etl_silver_test2", ts - 2 * one_day)
-    _make_and_fill_timestamps(db, "_etl_test_raw", ts - 2 * one_day)
+    _make_and_fill_timestamps(db, "_new_events_silver_test2", ts - 2 * one_day)
+    _make_and_fill_timestamps(db, "_unknown_test_raw", ts - 2 * one_day)
 
     mock_ppss = Mock()
 
@@ -203,11 +203,13 @@ def test_do_lake_etl_drop(tmpdir, caplog):
 
     assert "drop table _temp_bronze_test1 starting at 1609459200000" in caplog.text
     assert "rows before: 5" in caplog.text
-    assert "rows after: 2" in caplog.text
-    assert "drop table _etl_silver_test2 starting at 1609459200000" in caplog.text
+    assert "rows after: 3" in caplog.text
+    assert (
+        "drop table _new_events_silver_test2 starting at 1609459200000" in caplog.text
+    )
     assert "rows before: 5" in caplog.text
     assert "rows after: 3" in caplog.text
-    assert "skipping non-etl table _etl_test_raw" in caplog.text
+    assert "skipping non-etl table _unknown_test_raw" in caplog.text
     assert "truncated 5 rows from 2 tables" in caplog.text
 
 

diff --git a/pdr_backend/lake/csv_data_store.py b/pdr_backend/lake/csv_data_store.py
@@ -128,7 +128,7 @@ def __init__(self, base_path: str, table_name: str):
 
     @staticmethod
     def from_table(table, ppss):
-        return CSVDataStore(ppss.lake_ss.lake_dir, table.table_name)
+        return CSVDataStore(ppss.lake_ss.lake_dir, table._base_table_name)
 
     @enforce_types
     def _create_file_name(self, start_time: int, end_time: Optional[int]) -> str:
@@ -254,6 +254,7 @@ def _append_remaining_rows(
 
         remaining_data = data.slice(0, remaining_rows)
         last_file_path = self._get_last_file_path()
+
         last_file_data = pl.read_csv(last_file_path, schema=schema)
         last_file_data = last_file_data.vstack(remaining_data).rechunk()