1use anyhow::{anyhow, Context, Result};
2use arrow::array::{Array, BinaryArray, StringArray};
3use arrow::{array::RecordBatch, datatypes::*};
4use base64::{engine::general_purpose::STANDARD, Engine as _};
5use std::sync::Arc;
6mod deserialize;
7pub use deserialize::{deserialize_data, DynType, DynValue, ParamInput};
8mod arrow_converter;
9use arrow_converter::{to_arrow, to_arrow_dtype};
10
11#[derive(Debug, Clone)]
12pub struct InstructionSignature {
13 pub discriminator: Vec<u8>,
14 pub params: Vec<ParamInput>,
15 pub accounts_names: Vec<String>,
16}
17
18#[derive(Debug, Clone)]
19pub struct LogSignature {
20 pub params: Vec<ParamInput>,
21}
22
23#[cfg(feature = "pyo3")]
24impl<'py> pyo3::FromPyObject<'py> for InstructionSignature {
25 fn extract_bound(ob: &pyo3::Bound<'py, pyo3::PyAny>) -> pyo3::PyResult<Self> {
26 use pyo3::types::PyAnyMethods;
27 use pyo3::types::PyTypeMethods;
28
29 let discriminator_ob = ob.getattr("discriminator")?;
30
31 let discriminator_ob_type: String = discriminator_ob.get_type().name()?.to_string();
32 let discriminator = match discriminator_ob_type.as_str() {
33 "str" => {
34 let s: &str = discriminator_ob.extract()?;
35 hex_to_bytes(s).context("failed to decode hex")?
36 }
37 "bytes" => discriminator_ob.extract()?,
38 _ => return Err(anyhow!("unknown type: {}", discriminator_ob_type).into()),
39 };
40
41 let params = ob.getattr("params")?.extract::<Vec<ParamInput>>()?;
42 let accounts_names = ob.getattr("accounts_names")?.extract::<Vec<String>>()?;
43
44 Ok(InstructionSignature {
45 discriminator,
46 params,
47 accounts_names,
48 })
49 }
50}
51
52fn hex_to_bytes(hex_string: &str) -> Result<Vec<u8>> {
53 let hex_string = hex_string.strip_prefix("0x").unwrap_or(hex_string);
54 let hex_string = if hex_string.len() % 2 == 1 {
55 format!("0{}", hex_string)
56 } else {
57 hex_string.to_string()
58 };
59 let out = (0..hex_string.len())
60 .step_by(2)
61 .map(|i| {
62 u8::from_str_radix(&hex_string[i..i + 2], 16)
63 .context("failed to parse hexstring to bytes")
64 })
65 .collect::<Result<Vec<_>, _>>()?;
66
67 Ok(out)
68}
69
70#[cfg(feature = "pyo3")]
71impl<'py> pyo3::FromPyObject<'py> for LogSignature {
72 fn extract_bound(ob: &pyo3::Bound<'py, pyo3::PyAny>) -> pyo3::PyResult<Self> {
73 use pyo3::types::PyAnyMethods;
74
75 let params = ob.getattr("params")?.extract::<Vec<ParamInput>>()?;
76
77 Ok(LogSignature { params })
78 }
79}
80
81pub fn svm_decode_instructions(
82 signature: InstructionSignature,
83 batch: &RecordBatch,
84 allow_decode_fail: bool,
85) -> Result<RecordBatch> {
86 let data_col = batch.column_by_name("data").unwrap();
87 let data_array = data_col.as_any().downcast_ref::<BinaryArray>().unwrap();
88
89 let account_arrays: Vec<&BinaryArray> = (0..10)
90 .map(|i| {
91 let col_name = format!("a{}", i);
92 let col = batch.column_by_name(&col_name).unwrap();
93 col.as_any().downcast_ref::<BinaryArray>().unwrap()
94 })
95 .collect();
96
97 decode_instructions(signature, &account_arrays, data_array, allow_decode_fail)
98}
99
100pub fn decode_instructions(
101 signature: InstructionSignature,
102 accounts: &[&BinaryArray],
103 data: &BinaryArray,
104 allow_decode_fail: bool,
105) -> Result<RecordBatch> {
106 let num_params = signature.params.len();
107
108 let mut decoded_params_vec: Vec<Vec<Option<DynValue>>> =
109 (0..num_params).map(|_| Vec::new()).collect();
110
111 for row_idx in 0..data.len() {
112 if data.is_null(row_idx) {
113 if allow_decode_fail {
114 log::debug!("Instruction data is null in row {}", row_idx);
115 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
116 continue;
117 } else {
118 return Err(anyhow::anyhow!(
119 "Instruction data is null in row {}",
120 row_idx
121 ));
122 }
123 }
124
125 let instruction_data = data.value(row_idx).to_vec();
126 let data_result = match_discriminators(&instruction_data, &signature.discriminator);
127 let data = match data_result {
128 Ok(data) => data,
129 Err(e) if allow_decode_fail => {
130 log::debug!("Error matching discriminators in row {}: {:?}", row_idx, e);
131 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
132 continue;
133 }
134 Err(e) => {
135 return Err(anyhow::anyhow!(
136 "Error matching discriminators in row {}: {:?}",
137 row_idx,
138 e
139 ));
140 }
141 };
142
143 let decoded_ix_result = deserialize_data(&data, &signature.params);
144 let decoded_ix = match decoded_ix_result {
145 Ok(ix) => ix,
146 Err(e) if allow_decode_fail => {
147 log::debug!(
148 "Error deserializing instruction in row {}: {:?}",
149 row_idx,
150 e
151 );
152 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
153 continue;
154 }
155 Err(e) => {
156 return Err(anyhow::anyhow!(
157 "Error deserializing instruction in row {}: {:?}",
158 row_idx,
159 e
160 ));
161 }
162 };
163
164 for (i, value) in decoded_ix.into_iter().enumerate() {
165 decoded_params_vec[i].push(Some(value));
166 }
167 }
168
169 let data_arrays: Vec<Arc<dyn Array>> = decoded_params_vec
170 .iter()
171 .enumerate()
172 .map(|(i, v)| to_arrow(&signature.params[i].param_type, v.clone()).unwrap())
173 .collect::<Vec<_>>();
174
175 let data_fields = signature
176 .params
177 .iter()
178 .map(|p| Field::new(p.name.clone(), to_arrow_dtype(&p.param_type).unwrap(), true))
179 .collect::<Vec<_>>();
180
181 let acc_names_len = signature.accounts_names.len();
182
183 let mut accounts: Vec<Arc<dyn Array>> = accounts
184 .iter()
185 .map(|arr| {
186 let owned_array = arr.slice(0, arr.len());
187 owned_array as Arc<dyn Array>
188 })
189 .collect();
190
191 let mut acc_fields = Vec::new();
192 if acc_names_len < 10 {
193 let _ = accounts.split_off(acc_names_len);
194 for i in 0..acc_names_len {
195 let field = Field::new(signature.accounts_names[i].clone(), DataType::Binary, true);
196 acc_fields.push(field);
197 }
198 } else {
199 for i in 0..10 {
200 let field = Field::new(signature.accounts_names[i].clone(), DataType::Binary, true);
201 acc_fields.push(field);
202 }
203 }
204
205 let decoded_instructions_array = data_arrays.into_iter().chain(accounts).collect::<Vec<_>>();
206 let decoded_instructions_fields = data_fields
207 .into_iter()
208 .chain(acc_fields.clone())
209 .collect::<Vec<_>>();
210
211 let schema = Arc::new(Schema::new(decoded_instructions_fields));
212 let batch = RecordBatch::try_new(schema, decoded_instructions_array)
213 .context("Failed to create record batch from data arrays")
214 .unwrap();
215
216 Ok(batch)
217}
218
219pub fn svm_decode_logs(
220 signature: LogSignature,
221 batch: &RecordBatch,
222 allow_decode_fail: bool,
223) -> Result<RecordBatch> {
224 let message_col = batch.column_by_name("message").unwrap();
225 let data = message_col.as_any().downcast_ref::<StringArray>().unwrap();
226
227 decode_logs(signature, data, allow_decode_fail)
228}
229
230pub fn decode_logs(
231 signature: LogSignature,
232 data: &StringArray,
233 allow_decode_fail: bool,
234) -> Result<RecordBatch> {
235 let num_params = signature.params.len();
236
237 let mut decoded_params_vec: Vec<Vec<Option<DynValue>>> =
238 (0..num_params).map(|_| Vec::new()).collect();
239
240 for row_idx in 0..data.len() {
241 if data.is_null(row_idx) {
242 if allow_decode_fail {
243 log::debug!("Log data is null in row {}", row_idx);
244 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
245 continue;
246 } else {
247 return Err(anyhow::anyhow!("Log data is null in row {}", row_idx));
248 }
249 }
250
251 let log_data = data.value(row_idx);
252 let log_data = STANDARD.decode(log_data);
253 let log_data = match log_data {
254 Ok(log_data) => log_data,
255 Err(e) if allow_decode_fail => {
256 log::debug!(
257 "Error base 64 decoding log data in row {}: {:?}",
258 row_idx,
259 e
260 );
261 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
262 continue;
263 }
264 Err(e) => {
265 return Err(anyhow::anyhow!(
266 "Error base 64 decoding log data in row {}: {:?}",
267 row_idx,
268 e
269 ));
270 }
271 };
272
273 let decoded_log_result = deserialize_data(&log_data, &signature.params);
274 let decoded_log = match decoded_log_result {
275 Ok(log) => log,
276 Err(e) if allow_decode_fail => {
277 log::debug!("Error deserializing log in row {}: {:?}", row_idx, e);
278 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
279 continue;
280 }
281 Err(e) => {
282 return Err(anyhow::anyhow!(
283 "Error deserializing log in row {}: {:?}",
284 row_idx,
285 e
286 ));
287 }
288 };
289
290 for (i, value) in decoded_log.into_iter().enumerate() {
291 decoded_params_vec[i].push(Some(value));
292 }
293 }
294
295 let data_arrays: Vec<Arc<dyn Array>> = decoded_params_vec
296 .iter()
297 .enumerate()
298 .map(|(i, v)| to_arrow(&signature.params[i].param_type, v.clone()).unwrap())
299 .collect::<Vec<_>>();
300
301 let data_fields = signature
302 .params
303 .iter()
304 .map(|p| Field::new(p.name.clone(), to_arrow_dtype(&p.param_type).unwrap(), true))
305 .collect::<Vec<_>>();
306
307 let schema = Arc::new(Schema::new(data_fields));
308 let batch = RecordBatch::try_new(schema, data_arrays)
309 .context("Failed to create record batch from data arrays")
310 .unwrap();
311
312 Ok(batch)
313}
314
315pub fn match_discriminators(instr_data: &[u8], discriminator: &[u8]) -> Result<Vec<u8>> {
316 let discriminator_len = discriminator.len();
317 if instr_data.len() < discriminator_len {
318 return Err(anyhow::anyhow!(
319 "Instruction data is too short to contain discriminator. Expected at least {} bytes, got {} bytes",
320 discriminator_len,
321 instr_data.len()
322 ));
323 }
324 let disc = &instr_data[..discriminator_len].to_vec();
325 let ix_data = &instr_data[discriminator_len..];
326 if !disc.eq(discriminator) {
327 return Err(anyhow::anyhow!(
328 "Instruction data discriminator doesn't match signature discriminator"
329 ));
330 }
331 Ok(ix_data.to_vec())
332}
333
334pub fn instruction_signature_to_arrow_schema(signature: &InstructionSignature) -> Result<Schema> {
335 let mut fields = Vec::new();
336
337 for param in &signature.params {
338 let field = Field::new(
339 param.name.clone(),
340 to_arrow_dtype(¶m.param_type).unwrap(),
341 true,
342 );
343 fields.push(field);
344 }
345
346 for account in &signature.accounts_names {
347 let field = Field::new(account.clone(), DataType::Binary, true);
348 fields.push(field);
349 }
350
351 Ok(Schema::new(fields))
352}
353
354#[cfg(test)]
355mod tests {
356 use super::*;
357 use crate::deserialize::{DynType, ParamInput};
358 use std::fs::File;
359
360 #[test]
361 #[ignore]
362 fn test_instructions_with_real_data() {
363 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
364
365 let builder = ParquetRecordBatchReaderBuilder::try_new(
366 File::open("instruction_exemple.parquet").unwrap(),
367 )
368 .unwrap();
369 let mut reader = builder.build().unwrap();
370 let instructions = reader.next().unwrap().unwrap();
371 let ix_signature = InstructionSignature {
372 discriminator: vec![229, 23, 203, 151, 122, 227, 173, 42],
414 params: vec![
415 ParamInput {
416 name: "RoutePlan".to_string(),
417 param_type: DynType::Array(Box::new(DynType::Struct(vec![
418 (
419 "Swap".to_string(),
420 DynType::Enum(vec![
421 ("Saber".to_string(), None),
422 ("SaberAddDecimalsDeposit".to_string(), None),
423 ("SaberAddDecimalsWithdraw".to_string(), None),
424 ("TokenSwap".to_string(), None),
425 ("Sencha".to_string(), None),
426 ("Step".to_string(), None),
427 ("Cropper".to_string(), None),
428 ("Raydium".to_string(), None),
429 (
430 "Crema".to_string(),
431 Some(DynType::Struct(vec![(
432 "a_to_b".to_string(),
433 DynType::Bool,
434 )])),
435 ),
436 ("Lifinity".to_string(), None),
437 ("Mercurial".to_string(), None),
438 ("Cykura".to_string(), None),
439 (
440 "Serum".to_string(),
441 Some(DynType::Struct(vec![(
442 "side".to_string(),
443 DynType::Enum(vec![
444 ("Bid".to_string(), None),
445 ("Ask".to_string(), None),
446 ]),
447 )])),
448 ),
449 ("MarinadeDeposit".to_string(), None),
450 ("MarinadeUnstake".to_string(), None),
451 (
452 "Aldrin".to_string(),
453 Some(DynType::Struct(vec![(
454 "side".to_string(),
455 DynType::Enum(vec![
456 ("Bid".to_string(), None),
457 ("Ask".to_string(), None),
458 ]),
459 )])),
460 ),
461 (
462 "AldrinV2".to_string(),
463 Some(DynType::Struct(vec![(
464 "side".to_string(),
465 DynType::Enum(vec![
466 ("Bid".to_string(), None),
467 ("Ask".to_string(), None),
468 ]),
469 )])),
470 ),
471 (
472 "Whirlpool".to_string(),
473 Some(DynType::Struct(vec![(
474 "a_to_b".to_string(),
475 DynType::Bool,
476 )])),
477 ),
478 (
479 "Invariant".to_string(),
480 Some(DynType::Struct(vec![(
481 "x_to_y".to_string(),
482 DynType::Bool,
483 )])),
484 ),
485 ("Meteora".to_string(), None),
486 ("GooseFX".to_string(), None),
487 (
488 "DeltaFi".to_string(),
489 Some(DynType::Struct(vec![(
490 "stable".to_string(),
491 DynType::Bool,
492 )])),
493 ),
494 ("Balansol".to_string(), None),
495 (
496 "MarcoPolo".to_string(),
497 Some(DynType::Struct(vec![(
498 "x_to_y".to_string(),
499 DynType::Bool,
500 )])),
501 ),
502 (
503 "Dradex".to_string(),
504 Some(DynType::Struct(vec![(
505 "side".to_string(),
506 DynType::Enum(vec![
507 ("Bid".to_string(), None),
508 ("Ask".to_string(), None),
509 ]),
510 )])),
511 ),
512 ("LifinityV2".to_string(), None),
513 ("RaydiumClmm".to_string(), None),
514 (
515 "Openbook".to_string(),
516 Some(DynType::Struct(vec![(
517 "side".to_string(),
518 DynType::Enum(vec![
519 ("Bid".to_string(), None),
520 ("Ask".to_string(), None),
521 ]),
522 )])),
523 ),
524 (
525 "Phoenix".to_string(),
526 Some(DynType::Struct(vec![(
527 "side".to_string(),
528 DynType::Enum(vec![
529 ("Bid".to_string(), None),
530 ("Ask".to_string(), None),
531 ]),
532 )])),
533 ),
534 (
535 "Symmetry".to_string(),
536 Some(DynType::Struct(vec![
537 ("from_token_id".to_string(), DynType::U64),
538 ("to_token_id".to_string(), DynType::U64),
539 ])),
540 ),
541 ("TokenSwapV2".to_string(), None),
542 ("HeliumTreasuryManagementRedeemV0".to_string(), None),
543 ("StakeDexStakeWrappedSol".to_string(), None),
544 (
545 "StakeDexSwapViaStake".to_string(),
546 Some(DynType::Struct(vec![(
547 "bridge_stake_seed".to_string(),
548 DynType::U32,
549 )])),
550 ),
551 ("GooseFXV2".to_string(), None),
552 ("Perps".to_string(), None),
553 ("PerpsAddLiquidity".to_string(), None),
554 ("PerpsRemoveLiquidity".to_string(), None),
555 ("MeteoraDlmm".to_string(), None),
556 (
557 "OpenBookV2".to_string(),
558 Some(DynType::Struct(vec![(
559 "side".to_string(),
560 DynType::Enum(vec![
561 ("Bid".to_string(), None),
562 ("Ask".to_string(), None),
563 ]),
564 )])),
565 ),
566 ("RaydiumClmmV2".to_string(), None),
567 (
568 "StakeDexPrefundWithdrawStakeAndDepositStake".to_string(),
569 Some(DynType::Struct(vec![(
570 "bridge_stake_seed".to_string(),
571 DynType::U32,
572 )])),
573 ),
574 (
575 "Clone".to_string(),
576 Some(DynType::Struct(vec![
577 ("pool_index".to_string(), DynType::U8),
578 ("quantity_is_input".to_string(), DynType::Bool),
579 ("quantity_is_collateral".to_string(), DynType::Bool),
580 ])),
581 ),
582 (
583 "SanctumS".to_string(),
584 Some(DynType::Struct(vec![
585 ("src_lst_value_calc_accs".to_string(), DynType::U8),
586 ("dst_lst_value_calc_accs".to_string(), DynType::U8),
587 ("src_lst_index".to_string(), DynType::U32),
588 ("dst_lst_index".to_string(), DynType::U32),
589 ])),
590 ),
591 (
592 "SanctumSAddLiquidity".to_string(),
593 Some(DynType::Struct(vec![
594 ("lst_value_calc_accs".to_string(), DynType::U8),
595 ("lst_index".to_string(), DynType::U32),
596 ])),
597 ),
598 (
599 "SanctumSRemoveLiquidity".to_string(),
600 Some(DynType::Struct(vec![
601 ("lst_value_calc_accs".to_string(), DynType::U8),
602 ("lst_index".to_string(), DynType::U32),
603 ])),
604 ),
605 ("RaydiumCP".to_string(), None),
606 (
607 "WhirlpoolSwapV2".to_string(),
608 Some(DynType::Struct(vec![
609 ("a_to_b".to_string(), DynType::Bool),
610 (
611 "remaining_accounts_info".to_string(),
612 DynType::Struct(vec![(
613 "slices".to_string(),
614 DynType::Array(Box::new(DynType::Struct(vec![(
615 "remaining_accounts_slice".to_string(),
616 DynType::Struct(vec![
617 ("accounts_type".to_string(), DynType::U8),
618 ("length".to_string(), DynType::U8),
619 ]),
620 )]))),
621 )]),
622 ),
623 ])),
624 ),
625 ("OneIntro".to_string(), None),
626 ("PumpdotfunWrappedBuy".to_string(), None),
627 ("PumpdotfunWrappedSell".to_string(), None),
628 ("PerpsV2".to_string(), None),
629 ("PerpsV2AddLiquidity".to_string(), None),
630 ("PerpsV2RemoveLiquidity".to_string(), None),
631 ("MoonshotWrappedBuy".to_string(), None),
632 ("MoonshotWrappedSell".to_string(), None),
633 ("StabbleStableSwap".to_string(), None),
634 ("StabbleWeightedSwap".to_string(), None),
635 (
636 "Obric".to_string(),
637 Some(DynType::Struct(vec![(
638 "x_to_y".to_string(),
639 DynType::Bool,
640 )])),
641 ),
642 ("FoxBuyFromEstimatedCost".to_string(), None),
643 (
644 "FoxClaimPartial".to_string(),
645 Some(DynType::Struct(vec![(
646 "is_y".to_string(),
647 DynType::Bool,
648 )])),
649 ),
650 (
651 "SolFi".to_string(),
652 Some(DynType::Struct(vec![(
653 "is_quote_to_base".to_string(),
654 DynType::Bool,
655 )])),
656 ),
657 ("SolayerDelegateNoInit".to_string(), None),
658 ("SolayerUndelegateNoInit".to_string(), None),
659 (
660 "TokenMill".to_string(),
661 Some(DynType::Struct(vec![(
662 "side".to_string(),
663 DynType::Enum(vec![
664 ("Bid".to_string(), None),
665 ("Ask".to_string(), None),
666 ]),
667 )])),
668 ),
669 ("DaosFunBuy".to_string(), None),
670 ("DaosFunSell".to_string(), None),
671 ("ZeroFi".to_string(), None),
672 ("StakeDexWithdrawWrappedSol".to_string(), None),
673 ("VirtualsBuy".to_string(), None),
674 ("VirtualsSell".to_string(), None),
675 (
676 "Peren".to_string(),
677 Some(DynType::Struct(vec![
678 ("in_index".to_string(), DynType::U8),
679 ("out_index".to_string(), DynType::U8),
680 ])),
681 ),
682 ("PumpdotfunAmmBuy".to_string(), None),
683 ("PumpdotfunAmmSell".to_string(), None),
684 ("Gamma".to_string(), None),
685 ]),
686 ),
687 ("Percent".to_string(), DynType::U8),
688 ("InputIndex".to_string(), DynType::U8),
689 ("OutputIndex".to_string(), DynType::U8),
690 ]))),
691 },
692 ParamInput {
693 name: "InAmount".to_string(),
694 param_type: DynType::U64,
695 },
696 ParamInput {
697 name: "QuotedOutAmount".to_string(),
698 param_type: DynType::U64,
699 },
700 ParamInput {
701 name: "SlippageBps".to_string(),
702 param_type: DynType::U16,
703 },
704 ParamInput {
705 name: "PlatformFeeBps".to_string(),
706 param_type: DynType::U8,
707 },
708 ],
709 accounts_names: vec![
710 "TokenProgram".to_string(),
711 "UserTransferAuthority".to_string(),
712 "UserSourceTokenAccount".to_string(),
713 "UserDestinationTokenAccount".to_string(),
714 "DestinationTokenAccount".to_string(),
715 "PlatformFeeAccount".to_string(),
716 "EventAuthority".to_string(),
717 "Program".to_string(),
718 ],
719 };
720
721 let result = svm_decode_instructions(ix_signature, &instructions, true)
722 .context("decode failed")
723 .unwrap();
724
725 let mut file = File::create("decoded_instructions.parquet").unwrap();
727 let mut writer =
728 parquet::arrow::ArrowWriter::try_new(&mut file, result.schema(), None).unwrap();
729 writer.write(&result).unwrap();
730 writer.close().unwrap();
731 }
732
733 #[test]
734 #[ignore]
735 fn test_decode_logs_with_real_data() {
736 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
737
738 let builder =
739 ParquetRecordBatchReaderBuilder::try_new(File::open("logs.parquet").unwrap()).unwrap();
740 let mut reader = builder.build().unwrap();
741 let logs = reader.next().unwrap().unwrap();
742
743 let signature = LogSignature {
744 params: vec![
745 ParamInput {
746 name: "whirlpool".to_string(),
747 param_type: DynType::FixedArray(Box::new(DynType::U8), 32),
748 },
749 ParamInput {
750 name: "a_to_b".to_string(),
751 param_type: DynType::Bool,
752 },
753 ParamInput {
754 name: "pre_sqrt_price".to_string(),
755 param_type: DynType::U128,
756 },
757 ParamInput {
758 name: "post_sqrt_price".to_string(),
759 param_type: DynType::U128,
760 },
761 ParamInput {
762 name: "x".to_string(),
763 param_type: DynType::U64,
764 },
765 ParamInput {
766 name: "input_amount".to_string(),
767 param_type: DynType::U64,
768 },
769 ParamInput {
770 name: "output_amount".to_string(),
771 param_type: DynType::U64,
772 },
773 ParamInput {
774 name: "input_transfer_fee".to_string(),
775 param_type: DynType::U64,
776 },
777 ParamInput {
778 name: "output_transfer_fee".to_string(),
779 param_type: DynType::U64,
780 },
781 ParamInput {
782 name: "lp_fee".to_string(),
783 param_type: DynType::U64,
784 },
785 ParamInput {
786 name: "protocol_fee".to_string(),
787 param_type: DynType::U64,
788 },
789 ],
790 };
791
792 let result = svm_decode_logs(signature, &logs, true)
793 .context("decode failed")
794 .unwrap();
795
796 let mut file = File::create("decoded_logs.parquet").unwrap();
798 let mut writer =
799 parquet::arrow::ArrowWriter::try_new(&mut file, result.schema(), None).unwrap();
800 writer.write(&result).unwrap();
801 writer.close().unwrap();
802 }
803
804 #[test]
805 #[ignore]
806 fn test_instruction_signature_to_arrow_schema() {
807 let signature = InstructionSignature {
809 discriminator: vec![],
810 params: vec![
811 ParamInput {
812 name: "amount".to_string(),
813 param_type: DynType::U64,
814 },
815 ParamInput {
816 name: "is_valid".to_string(),
817 param_type: DynType::Bool,
818 },
819 ParamInput {
820 name: "amm".to_string(),
821 param_type: DynType::FixedArray(Box::new(DynType::U8), 32),
822 },
823 ],
824 accounts_names: vec!["source".to_string(), "destination".to_string()],
825 };
826
827 let schema = instruction_signature_to_arrow_schema(&signature).unwrap();
829
830 assert_eq!(schema.fields().len(), 5); let amount_field = schema.field_with_name("amount").unwrap();
835 assert_eq!(amount_field.name(), "amount");
836 assert!(amount_field.is_nullable());
837
838 let is_valid_field = schema.field_with_name("is_valid").unwrap();
839 assert_eq!(is_valid_field.name(), "is_valid");
840 assert!(is_valid_field.is_nullable());
841
842 let amm_field = schema.field_with_name("amm").unwrap();
843 assert_eq!(amm_field.name(), "amm");
844 assert!(amm_field.is_nullable());
845
846 let source_field = schema.field_with_name("source").unwrap();
848 assert_eq!(source_field.name(), "source");
849 assert_eq!(source_field.data_type(), &DataType::Binary);
850 assert!(source_field.is_nullable());
851
852 let dest_field = schema.field_with_name("destination").unwrap();
853 assert_eq!(dest_field.name(), "destination");
854 assert_eq!(dest_field.data_type(), &DataType::Binary);
855 assert!(dest_field.is_nullable());
856 }
857}