Address comments

cmd/export_assets.go

@@ -64,7 +64,9 @@ var assetsCmd = &cobra.Command{
 			}
 			totalNumBytes += numBytes
 
-			transformedAssets = append(transformedAssets, transformed)
+			if commonArgs.WriteParquet {
+				transformedAssets = append(transformedAssets, transformed)
+			}
 		}
 
 		outFile.Close()

cmd/export_contract_events.go

@@ -47,7 +47,9 @@ var contractEventsCmd = &cobra.Command{
 					continue
 				}
 
-				transformedEvents = append(transformedEvents, contractEvent)
+				if commonArgs.WriteParquet {
+					transformedEvents = append(transformedEvents, contractEvent)
+				}
 			}
 
 		}

cmd/export_effects.go

@@ -48,7 +48,9 @@ var effectsCmd = &cobra.Command{
 				}
 				totalNumBytes += numBytes
 
-				transformedEffects = append(transformedEffects, transformed)
+				if commonArgs.WriteParquet {
+					transformedEffects = append(transformedEffects, transformed)
+				}
 			}
 		}
 

cmd/export_ledger_entry_changes.go

@@ -302,46 +302,51 @@ func exportTransformedData(
 			if err != nil {
 				return err
 			}
-			switch v := o.(type) {
-			case transform.AccountOutput:
-				transformedResource = append(transformedResource, v)
-				parquetSchema = new(transform.AccountOutputParquet)
-				skip = false
-			case transform.AccountSignerOutput:
-				transformedResource = append(transformedResource, v)
-				parquetSchema = new(transform.AccountSignerOutputParquet)
-				skip = false
-			case transform.ClaimableBalanceOutput:
-				// Skip
-				skip = true
-			case transform.ConfigSettingOutput:
-				transformedResource = append(transformedResource, v)
-				parquetSchema = new(transform.ConfigSettingOutputParquet)
-				skip = false
-			case transform.ContractCodeOutput:
-				transformedResource = append(transformedResource, v)
-				parquetSchema = new(transform.ContractCodeOutputParquet)
-				skip = false
-			case transform.ContractDataOutput:
-				transformedResource = append(transformedResource, v)
-				parquetSchema = new(transform.ContractDataOutputParquet)
-				skip = false
-			case transform.PoolOutput:
-				transformedResource = append(transformedResource, v)
-				parquetSchema = new(transform.PoolOutputParquet)
-				skip = false
-			case transform.OfferOutput:
-				transformedResource = append(transformedResource, v)
-				parquetSchema = new(transform.OfferOutputParquet)
-				skip = false
-			case transform.TrustlineOutput:
-				transformedResource = append(transformedResource, v)
-				parquetSchema = new(transform.TrustlineOutputParquet)
-				skip = false
-			case transform.TtlOutput:
-				transformedResource = append(transformedResource, v)
-				parquetSchema = new(transform.TtlOutputParquet)
-				skip = false
+
+			if WriteParquet {
+				switch v := o.(type) {
+				case transform.AccountOutput:
+					transformedResource = append(transformedResource, v)
+					parquetSchema = new(transform.AccountOutputParquet)
+					skip = false
+				case transform.AccountSignerOutput:
+					transformedResource = append(transformedResource, v)
+					parquetSchema = new(transform.AccountSignerOutputParquet)
+					skip = false
+				case transform.ClaimableBalanceOutput:
+					// Skipping ClaimableBalanceOutputParquet because it is not needed in the current scope of work
+					// Note that ClaimableBalanceOutputParquet uses nested structs that will need to be handled
+					// for parquet conversio
+					skip = true
+				case transform.ConfigSettingOutput:
+					transformedResource = append(transformedResource, v)
+					parquetSchema = new(transform.ConfigSettingOutputParquet)
+					skip = false
+				case transform.ContractCodeOutput:
+					transformedResource = append(transformedResource, v)
+					parquetSchema = new(transform.ContractCodeOutputParquet)
+					skip = false
+				case transform.ContractDataOutput:
+					transformedResource = append(transformedResource, v)
+					parquetSchema = new(transform.ContractDataOutputParquet)
+					skip = false
+				case transform.PoolOutput:
+					transformedResource = append(transformedResource, v)
+					parquetSchema = new(transform.PoolOutputParquet)
+					skip = false
+				case transform.OfferOutput:
+					transformedResource = append(transformedResource, v)
+					parquetSchema = new(transform.OfferOutputParquet)
+					skip = false
+				case transform.TrustlineOutput:
+					transformedResource = append(transformedResource, v)
+					parquetSchema = new(transform.TrustlineOutputParquet)
+					skip = false
+				case transform.TtlOutput:
+					transformedResource = append(transformedResource, v)
+					parquetSchema = new(transform.TtlOutputParquet)
+					skip = false
+				}
 			}
 		}
 

cmd/export_ledgers.go

@@ -55,7 +55,9 @@ var ledgersCmd = &cobra.Command{
 			}
 			totalNumBytes += numBytes
 
-			transformedLedgers = append(transformedLedgers, transformed)
+			if commonArgs.WriteParquet {
+				transformedLedgers = append(transformedLedgers, transformed)
+			}
 		}
 
 		outFile.Close()

cmd/export_operations.go

@@ -48,7 +48,9 @@ var operationsCmd = &cobra.Command{
 			}
 			totalNumBytes += numBytes
 
-			transformedOps = append(transformedOps, transformed)
+			if commonArgs.WriteParquet {
+				transformedOps = append(transformedOps, transformed)
+			}
 		}
 
 		outFile.Close()

cmd/export_trades.go

@@ -52,7 +52,9 @@ var tradesCmd = &cobra.Command{
 				}
 				totalNumBytes += numBytes
 
-				transformedTrades = append(transformedTrades, transformed)
+				if commonArgs.WriteParquet {
+					transformedTrades = append(transformedTrades, transformed)
+				}
 			}
 		}
 

cmd/export_transactions.go

@@ -48,7 +48,9 @@ var transactionsCmd = &cobra.Command{
 			}
 			totalNumBytes += numBytes
 
-			transformedTransaction = append(transformedTransaction, transformed)
+			if commonArgs.WriteParquet {
+				transformedTransaction = append(transformedTransaction, transformed)
+			}
 		}
 
 		outFile.Close()

internal/transform/parquet_converter.go

@@ -1,5 +1,10 @@
 // This file includes interfaces and functions for converting data structures/schemas
 // to the appropriate parquet data structure/schema.
+//
+// Note that uint32 data types need to be converted to int64 due to restrictions
+// from the parquet-go package. Conversion is to int64 due to the possible loss of
+// data in the conversion from uint32 -> int32.
+// This applies to all the ToParquet() functions in this file.
 
 package transform
 
@@ -21,6 +26,9 @@ func toJSONString(v interface{}) string {
 
 func (lo LedgerOutput) ToParquet() interface{} {
 	return LedgerOutputParquet{
+		// Note that uint32 data types need to be converted to int64 due to restrictions
+		// from the parquet-go package. Conversion is to int64 due to the possible loss of
+		// data in the conversion from uint32 -> int32.
 		Sequence:                   int64(lo.Sequence),
 		LedgerHash:                 lo.LedgerHash,
 		PreviousLedgerHash:         lo.PreviousLedgerHash,

internal/transform/schema_parquet.go

@@ -122,7 +122,9 @@ type OperationOutputParquet struct {
 	LedgerSequence      int64  `parquet:"name=ledger_sequence, type=INT64, convertedtype=INT64, convertedtype=UINT_64"`
 }
 
-//// Skipping ClaimableBalanceOutputParquet because it is not needed
+//// Skipping ClaimableBalanceOutputParquet because it is not needed in the current scope of work
+//// Note that ClaimableBalanceOutputParquet uses nested structs that will need to be handled
+//// for parquet conversion
 //type ClaimableBalanceOutputParquet struct {
 //}
 

internal/utils/main.go

@@ -240,7 +240,7 @@ func AddCommonFlags(flags *pflag.FlagSet) {
 	flags.Uint32("num-workers", 10, "Number of workers to spawn that read txmeta files from the datastore.")
 	flags.Uint32("retry-limit", 3, "Datastore GetLedger retry limit.")
 	flags.Uint32("retry-wait", 5, "Time in seconds to wait for GetLedger retry.")
-	flags.Bool("write-parquet", true, "If set, write output as parquet files.")
+	flags.Bool("write-parquet", false, "If set, write output as parquet files.")
 }
 
 // AddArchiveFlags adds the history archive specific flags: output, and limit