1use anyhow::{anyhow, Context, Result};
2use arrow::array::{
3 builder, Array, BinaryArray, GenericBinaryArray, GenericListArray, GenericStringArray,
4 LargeBinaryArray, LargeStringArray, OffsetSizeTrait, StringArray,
5};
6use arrow::{array::RecordBatch, datatypes::*};
7use base64::{engine::general_purpose::STANDARD, Engine as _};
8use std::sync::Arc;
9mod deserialize;
10pub use deserialize::{deserialize_data, DynType, DynValue, ParamInput};
11mod arrow_converter;
12use arrow_converter::{to_arrow, to_arrow_dtype};
13
14#[derive(Debug, Clone)]
15pub struct InstructionSignature {
16 pub discriminator: Vec<u8>,
17 pub params: Vec<ParamInput>,
18 pub accounts_names: Vec<String>,
19}
20
21#[derive(Debug, Clone)]
22pub struct LogSignature {
23 pub params: Vec<ParamInput>,
24}
25
26#[cfg(feature = "pyo3")]
27impl<'py> pyo3::FromPyObject<'py> for InstructionSignature {
28 fn extract_bound(ob: &pyo3::Bound<'py, pyo3::PyAny>) -> pyo3::PyResult<Self> {
29 use pyo3::types::PyAnyMethods;
30 use pyo3::types::PyTypeMethods;
31
32 let discriminator_ob = ob.getattr("discriminator")?;
33
34 let discriminator_ob_type: String = discriminator_ob.get_type().name()?.to_string();
35 let discriminator = match discriminator_ob_type.as_str() {
36 "str" => {
37 let s: &str = discriminator_ob.extract()?;
38 hex_to_bytes(s).context("failed to decode hex")?
39 }
40 "bytes" => discriminator_ob.extract()?,
41 _ => return Err(anyhow!("unknown type: {}", discriminator_ob_type).into()),
42 };
43
44 let params = ob.getattr("params")?.extract::<Vec<ParamInput>>()?;
45 let accounts_names = ob.getattr("accounts_names")?.extract::<Vec<String>>()?;
46
47 Ok(InstructionSignature {
48 discriminator,
49 params,
50 accounts_names,
51 })
52 }
53}
54
55fn hex_to_bytes(hex_string: &str) -> Result<Vec<u8>> {
56 let hex_string = hex_string.strip_prefix("0x").unwrap_or(hex_string);
57 let hex_string = if hex_string.len() % 2 == 1 {
58 format!("0{}", hex_string)
59 } else {
60 hex_string.to_string()
61 };
62 let out = (0..hex_string.len())
63 .step_by(2)
64 .map(|i| {
65 u8::from_str_radix(&hex_string[i..i + 2], 16)
66 .context("failed to parse hexstring to bytes")
67 })
68 .collect::<Result<Vec<_>, _>>()?;
69
70 Ok(out)
71}
72
73#[cfg(feature = "pyo3")]
74impl<'py> pyo3::FromPyObject<'py> for LogSignature {
75 fn extract_bound(ob: &pyo3::Bound<'py, pyo3::PyAny>) -> pyo3::PyResult<Self> {
76 use pyo3::types::PyAnyMethods;
77
78 let params = ob.getattr("params")?.extract::<Vec<ParamInput>>()?;
79
80 Ok(LogSignature { params })
81 }
82}
83
84fn unpack_rest_of_accounts<ListI: OffsetSizeTrait, InnerI: OffsetSizeTrait>(
85 num_acc: usize,
86 rest_of_acc: &GenericListArray<ListI>,
87 account_arrays: &mut Vec<BinaryArray>,
88) -> Result<()> {
89 let data_size = rest_of_acc.len() * 32;
90
91 for acc_arr in rest_of_acc.iter().flatten() {
92 if acc_arr.len() < num_acc {
93 return Err(anyhow!(
94 "expected rest_of_accounts to have at least {} addresses but it has {}",
95 num_acc,
96 acc_arr.len()
97 ));
98 }
99 }
100
101 for i in 0..num_acc {
102 let mut builder = builder::BinaryBuilder::with_capacity(rest_of_acc.len(), data_size);
103
104 for acc_arr in rest_of_acc.iter() {
105 let acc_arr = match acc_arr {
106 Some(a) => a,
107 None => {
108 builder.append_null();
109 continue;
110 }
111 };
112
113 let arr = acc_arr
114 .as_any()
115 .downcast_ref::<GenericBinaryArray<InnerI>>()
116 .unwrap();
117 if !arr.is_null(i) {
118 builder.append_value(arr.value(i));
119 } else {
120 builder.append_null();
121 }
122 }
123
124 account_arrays.push(builder.finish());
125 }
126
127 Ok(())
128}
129
130pub fn decode_instructions_batch(
131 signature: InstructionSignature,
132 batch: &RecordBatch,
133 allow_decode_fail: bool,
134) -> Result<RecordBatch> {
135 let mut account_arrays: Vec<BinaryArray> = Vec::with_capacity(20);
136
137 for i in 0..signature.accounts_names.len().min(10) {
138 let col_name = format!("a{}", i);
139 let col = batch
140 .column_by_name(&col_name)
141 .with_context(|| format!("account {} not found but was required", i))?;
142
143 if col.data_type() == &DataType::Binary {
144 account_arrays.push(col.as_any().downcast_ref::<BinaryArray>().unwrap().clone());
145 } else if col.data_type() == &DataType::LargeBinary {
146 account_arrays.push(
147 arrow::compute::cast(col, &DataType::Binary)
148 .unwrap()
149 .as_any()
150 .downcast_ref::<BinaryArray>()
151 .unwrap()
152 .clone(),
153 );
154 }
155 }
156
157 if signature.accounts_names.len() > 10 {
158 let rest_of_acc = batch
159 .column_by_name("rest_of_accounts")
160 .context("rest_of_accounts column not found in instructions batch")?;
161
162 let num_acc = signature.accounts_names.len() - 10;
163 if rest_of_acc.data_type() == &DataType::new_list(DataType::Binary, true) {
164 unpack_rest_of_accounts::<i32, i32>(
165 num_acc,
166 rest_of_acc.as_any().downcast_ref().unwrap(),
167 &mut account_arrays,
168 )
169 .context("unpack rest_of_accounts column")?;
170 } else if rest_of_acc.data_type() == &DataType::new_list(DataType::LargeBinary, true) {
171 unpack_rest_of_accounts::<i32, i64>(
172 num_acc,
173 rest_of_acc.as_any().downcast_ref().unwrap(),
174 &mut account_arrays,
175 )
176 .context("unpack rest_of_accounts column")?;
177 } else if rest_of_acc.data_type() == &DataType::new_large_list(DataType::Binary, true) {
178 unpack_rest_of_accounts::<i64, i32>(
179 num_acc,
180 rest_of_acc.as_any().downcast_ref().unwrap(),
181 &mut account_arrays,
182 )
183 .context("unpack rest_of_accounts column")?;
184 } else if rest_of_acc.data_type() == &DataType::new_large_list(DataType::LargeBinary, true)
185 {
186 unpack_rest_of_accounts::<i64, i64>(
187 num_acc,
188 rest_of_acc.as_any().downcast_ref().unwrap(),
189 &mut account_arrays,
190 )
191 .context("unpack rest_of_accounts column")?;
192 }
193 }
194
195 let data_col = batch
196 .column_by_name("data")
197 .context("data column not found in instructions batch")?;
198
199 if data_col.data_type() == &DataType::Binary {
200 decode_instructions(
201 signature,
202 &account_arrays,
203 data_col.as_any().downcast_ref::<BinaryArray>().unwrap(),
204 allow_decode_fail,
205 )
206 } else if data_col.data_type() == &DataType::LargeBinary {
207 decode_instructions(
208 signature,
209 &account_arrays,
210 data_col
211 .as_any()
212 .downcast_ref::<LargeBinaryArray>()
213 .unwrap(),
214 allow_decode_fail,
215 )
216 } else {
217 Err(anyhow!(
218 "expected the data column to be Binary or LargeBinary"
219 ))
220 }
221}
222
223pub fn decode_instructions<I: OffsetSizeTrait>(
224 signature: InstructionSignature,
225 accounts: &[BinaryArray],
226 data: &GenericBinaryArray<I>,
227 allow_decode_fail: bool,
228) -> Result<RecordBatch> {
229 let num_params = signature.params.len();
230
231 let mut decoded_params_vec: Vec<Vec<Option<DynValue>>> =
232 (0..num_params).map(|_| Vec::new()).collect();
233
234 for row_idx in 0..data.len() {
235 if data.is_null(row_idx) {
236 if allow_decode_fail {
237 log::debug!("Instruction data is null in row {}", row_idx);
238 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
239 continue;
240 } else {
241 return Err(anyhow::anyhow!(
242 "Instruction data is null in row {}",
243 row_idx
244 ));
245 }
246 }
247
248 let instruction_data = data.value(row_idx).to_vec();
249 let data_result = match_discriminators(&instruction_data, &signature.discriminator);
250 let data = match data_result {
251 Ok(data) => data,
252 Err(e) if allow_decode_fail => {
253 log::debug!("Error matching discriminators in row {}: {:?}", row_idx, e);
254 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
255 continue;
256 }
257 Err(e) => {
258 return Err(anyhow::anyhow!(
259 "Error matching discriminators in row {}: {:?}",
260 row_idx,
261 e
262 ));
263 }
264 };
265
266 let error_on_remanining = false;
272 let decoded_ix_result = deserialize_data(&data, &signature.params, error_on_remanining);
273 let decoded_ix = match decoded_ix_result {
274 Ok(ix) => ix,
275 Err(e) if allow_decode_fail => {
276 log::debug!(
277 "Error deserializing instruction in row {}: {:?}",
278 row_idx,
279 e
280 );
281 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
282 continue;
283 }
284 Err(e) => {
285 return Err(anyhow::anyhow!(
286 "Error deserializing instruction in row {}: {:?}",
287 row_idx,
288 e
289 ));
290 }
291 };
292
293 for (i, value) in decoded_ix.into_iter().enumerate() {
294 decoded_params_vec[i].push(Some(value));
295 }
296 }
297
298 let mut data_arrays: Vec<Arc<dyn Array>> = Vec::with_capacity(decoded_params_vec.len());
299 for (i, v) in decoded_params_vec.iter().enumerate() {
300 let array = to_arrow(&signature.params[i].param_type, v.clone())
301 .context("unable to convert instruction value to a arrow format value")?;
302 data_arrays.push(array);
303 }
304
305 let mut data_fields = Vec::with_capacity(signature.params.len());
306 for param in &signature.params {
307 let field = Field::new(
308 param.name.clone(),
309 to_arrow_dtype(¶m.param_type)
310 .context("unable to convert instruction param type to arrow dtype")?,
311 true,
312 );
313 data_fields.push(field);
314 }
315
316 let acc_names_len = signature.accounts_names.len();
317 let mut accounts_arrays = Vec::new();
318 let mut acc_fields = Vec::new();
319
320 for i in 0..acc_names_len {
321 let arr = accounts
322 .get(i)
323 .context(format!("Account a{} not found during decoding", i))?;
324
325 if arr.data_type() == &DataType::LargeBinary {
326 accounts_arrays.push(arrow::compute::cast(arr, &DataType::Binary).unwrap());
327 } else {
328 accounts_arrays.push(Arc::new(arr.clone()) as Arc<dyn Array>);
329 }
330
331 if signature.accounts_names[i].is_empty() {
332 let field = Field::new(format!("a{}", i), DataType::Binary, true);
333 acc_fields.push(field);
334 } else {
335 let field = Field::new(signature.accounts_names[i].clone(), DataType::Binary, true);
336 acc_fields.push(field);
337 }
338 }
339
340 let decoded_instructions_array = data_arrays
341 .into_iter()
342 .chain(accounts_arrays)
343 .collect::<Vec<_>>();
344 let decoded_instructions_fields = data_fields
345 .into_iter()
346 .chain(acc_fields.clone())
347 .collect::<Vec<_>>();
348
349 let schema = Arc::new(Schema::new(decoded_instructions_fields));
350 let batch = RecordBatch::try_new(schema, decoded_instructions_array)
351 .context("Failed to create record batch from data arrays")?;
352
353 Ok(batch)
354}
355
356pub fn decode_logs_batch(
357 signature: LogSignature,
358 batch: &RecordBatch,
359 allow_decode_fail: bool,
360) -> Result<RecordBatch> {
361 let message_col = batch
362 .column_by_name("message")
363 .context("message column not found in logs batch")?;
364
365 if message_col.data_type() == &DataType::Utf8 {
366 decode_logs(
367 signature,
368 message_col.as_any().downcast_ref::<StringArray>().unwrap(),
369 allow_decode_fail,
370 )
371 } else if message_col.data_type() == &DataType::LargeUtf8 {
372 decode_logs(
373 signature,
374 message_col
375 .as_any()
376 .downcast_ref::<LargeStringArray>()
377 .unwrap(),
378 allow_decode_fail,
379 )
380 } else {
381 Err(anyhow!("expected String or LargeString message column"))
382 }
383}
384
385pub fn decode_logs<I: OffsetSizeTrait>(
386 signature: LogSignature,
387 data: &GenericStringArray<I>,
388 allow_decode_fail: bool,
389) -> Result<RecordBatch> {
390 let num_params = signature.params.len();
391
392 let mut decoded_params_vec: Vec<Vec<Option<DynValue>>> =
393 (0..num_params).map(|_| Vec::new()).collect();
394
395 for row_idx in 0..data.len() {
396 if data.is_null(row_idx) {
397 if allow_decode_fail {
398 log::debug!("Log data is null in row {}", row_idx);
399 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
400 continue;
401 } else {
402 return Err(anyhow::anyhow!("Log data is null in row {}", row_idx));
403 }
404 }
405
406 let log_data = data.value(row_idx);
407 let log_data = STANDARD.decode(log_data);
408 let log_data = match log_data {
409 Ok(log_data) => log_data,
410 Err(e) if allow_decode_fail => {
411 log::debug!(
412 "Error base 64 decoding log data in row {}: {:?}",
413 row_idx,
414 e
415 );
416 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
417 continue;
418 }
419 Err(e) => {
420 return Err(anyhow::anyhow!(
421 "Error base 64 decoding log data in row {}: {:?}",
422 row_idx,
423 e
424 ));
425 }
426 };
427
428 let decoded_log_result = deserialize_data(&log_data, &signature.params, false);
429 let decoded_log = match decoded_log_result {
430 Ok(log) => log,
431 Err(e) if allow_decode_fail => {
432 log::debug!("Error deserializing log in row {}: {:?}", row_idx, e);
433 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
434 continue;
435 }
436 Err(e) => {
437 return Err(anyhow::anyhow!(
438 "Error deserializing log in row {}: {:?}",
439 row_idx,
440 e
441 ));
442 }
443 };
444
445 for (i, value) in decoded_log.into_iter().enumerate() {
446 decoded_params_vec[i].push(Some(value));
447 }
448 }
449
450 let mut data_arrays: Vec<Arc<dyn Array>> = Vec::with_capacity(decoded_params_vec.len());
451 for (i, v) in decoded_params_vec.iter().enumerate() {
452 let array = to_arrow(&signature.params[i].param_type, v.clone())
453 .context("unable to convert log value to a arrow format value")?;
454 data_arrays.push(array);
455 }
456
457 let mut data_fields = Vec::with_capacity(signature.params.len());
458 for param in &signature.params {
459 let field = Field::new(
460 param.name.clone(),
461 to_arrow_dtype(¶m.param_type)
462 .context("unable to convert log param type to arrow dtype")?,
463 true,
464 );
465 data_fields.push(field);
466 }
467
468 let schema = Arc::new(Schema::new(data_fields));
469 let batch = RecordBatch::try_new(schema, data_arrays)
470 .context("Failed to create record batch from data arrays")?;
471
472 Ok(batch)
473}
474
475pub fn match_discriminators(instr_data: &[u8], discriminator: &[u8]) -> Result<Vec<u8>> {
476 let discriminator_len = discriminator.len();
477 if instr_data.len() < discriminator_len {
478 return Err(anyhow::anyhow!(
479 "Instruction data is too short to contain discriminator. Expected at least {} bytes, got {} bytes",
480 discriminator_len,
481 instr_data.len()
482 ));
483 }
484 let disc = &instr_data[..discriminator_len].to_vec();
485 let ix_data = &instr_data[discriminator_len..];
486 if !disc.eq(discriminator) {
487 return Err(anyhow::anyhow!(
488 "Instruction data discriminator doesn't match signature discriminator"
489 ));
490 }
491 Ok(ix_data.to_vec())
492}
493
494pub fn instruction_signature_to_arrow_schema(signature: &InstructionSignature) -> Result<Schema> {
495 let mut fields = Vec::new();
496
497 for param in &signature.params {
498 let field = Field::new(
499 param.name.clone(),
500 to_arrow_dtype(¶m.param_type)
501 .context("unable to convert instruction param type to arrow dtype")?,
502 true,
503 );
504 fields.push(field);
505 }
506
507 for account in &signature.accounts_names {
508 let field = Field::new(account.clone(), DataType::Binary, true);
509 fields.push(field);
510 }
511
512 Ok(Schema::new(fields))
513}
514
515#[cfg(test)]
516mod tests {
517 use super::*;
518 use crate::deserialize::{DynType, ParamInput};
519 use std::fs::File;
520
521 #[test]
522 #[ignore]
523 fn test_instructions_with_real_data() {
524 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
525
526 let builder =
527 ParquetRecordBatchReaderBuilder::try_new(File::open("jup.parquet").unwrap()).unwrap();
528 let mut reader = builder.build().unwrap();
529 let instructions = reader.next().unwrap().unwrap();
530 let ix_signature = InstructionSignature {
531 discriminator: vec![229, 23, 203, 151, 122, 227, 173, 42],
573 params: vec![
574 ParamInput {
575 name: "RoutePlan".to_string(),
576 param_type: DynType::Array(Box::new(DynType::Struct(vec![
577 (
578 "Swap".to_string(),
579 DynType::Enum(vec![
580 ("Saber".to_string(), None),
581 ("SaberAddDecimalsDeposit".to_string(), None),
582 ("SaberAddDecimalsWithdraw".to_string(), None),
583 ("TokenSwap".to_string(), None),
584 ("Sencha".to_string(), None),
585 ("Step".to_string(), None),
586 ("Cropper".to_string(), None),
587 ("Raydium".to_string(), None),
588 (
589 "Crema".to_string(),
590 Some(DynType::Struct(vec![(
591 "a_to_b".to_string(),
592 DynType::Bool,
593 )])),
594 ),
595 ("Lifinity".to_string(), None),
596 ("Mercurial".to_string(), None),
597 ("Cykura".to_string(), None),
598 (
599 "Serum".to_string(),
600 Some(DynType::Struct(vec![(
601 "side".to_string(),
602 DynType::Enum(vec![
603 ("Bid".to_string(), None),
604 ("Ask".to_string(), None),
605 ]),
606 )])),
607 ),
608 ("MarinadeDeposit".to_string(), None),
609 ("MarinadeUnstake".to_string(), None),
610 (
611 "Aldrin".to_string(),
612 Some(DynType::Struct(vec![(
613 "side".to_string(),
614 DynType::Enum(vec![
615 ("Bid".to_string(), None),
616 ("Ask".to_string(), None),
617 ]),
618 )])),
619 ),
620 (
621 "AldrinV2".to_string(),
622 Some(DynType::Struct(vec![(
623 "side".to_string(),
624 DynType::Enum(vec![
625 ("Bid".to_string(), None),
626 ("Ask".to_string(), None),
627 ]),
628 )])),
629 ),
630 (
631 "Whirlpool".to_string(),
632 Some(DynType::Struct(vec![(
633 "a_to_b".to_string(),
634 DynType::Bool,
635 )])),
636 ),
637 (
638 "Invariant".to_string(),
639 Some(DynType::Struct(vec![(
640 "x_to_y".to_string(),
641 DynType::Bool,
642 )])),
643 ),
644 ("Meteora".to_string(), None),
645 ("GooseFX".to_string(), None),
646 (
647 "DeltaFi".to_string(),
648 Some(DynType::Struct(vec![(
649 "stable".to_string(),
650 DynType::Bool,
651 )])),
652 ),
653 ("Balansol".to_string(), None),
654 (
655 "MarcoPolo".to_string(),
656 Some(DynType::Struct(vec![(
657 "x_to_y".to_string(),
658 DynType::Bool,
659 )])),
660 ),
661 (
662 "Dradex".to_string(),
663 Some(DynType::Struct(vec![(
664 "side".to_string(),
665 DynType::Enum(vec![
666 ("Bid".to_string(), None),
667 ("Ask".to_string(), None),
668 ]),
669 )])),
670 ),
671 ("LifinityV2".to_string(), None),
672 ("RaydiumClmm".to_string(), None),
673 (
674 "Openbook".to_string(),
675 Some(DynType::Struct(vec![(
676 "side".to_string(),
677 DynType::Enum(vec![
678 ("Bid".to_string(), None),
679 ("Ask".to_string(), None),
680 ]),
681 )])),
682 ),
683 (
684 "Phoenix".to_string(),
685 Some(DynType::Struct(vec![(
686 "side".to_string(),
687 DynType::Enum(vec![
688 ("Bid".to_string(), None),
689 ("Ask".to_string(), None),
690 ]),
691 )])),
692 ),
693 (
694 "Symmetry".to_string(),
695 Some(DynType::Struct(vec![
696 ("from_token_id".to_string(), DynType::U64),
697 ("to_token_id".to_string(), DynType::U64),
698 ])),
699 ),
700 ("TokenSwapV2".to_string(), None),
701 ("HeliumTreasuryManagementRedeemV0".to_string(), None),
702 ("StakeDexStakeWrappedSol".to_string(), None),
703 (
704 "StakeDexSwapViaStake".to_string(),
705 Some(DynType::Struct(vec![(
706 "bridge_stake_seed".to_string(),
707 DynType::U32,
708 )])),
709 ),
710 ("GooseFXV2".to_string(), None),
711 ("Perps".to_string(), None),
712 ("PerpsAddLiquidity".to_string(), None),
713 ("PerpsRemoveLiquidity".to_string(), None),
714 ("MeteoraDlmm".to_string(), None),
715 (
716 "OpenBookV2".to_string(),
717 Some(DynType::Struct(vec![(
718 "side".to_string(),
719 DynType::Enum(vec![
720 ("Bid".to_string(), None),
721 ("Ask".to_string(), None),
722 ]),
723 )])),
724 ),
725 ("RaydiumClmmV2".to_string(), None),
726 (
727 "StakeDexPrefundWithdrawStakeAndDepositStake".to_string(),
728 Some(DynType::Struct(vec![(
729 "bridge_stake_seed".to_string(),
730 DynType::U32,
731 )])),
732 ),
733 (
734 "Clone".to_string(),
735 Some(DynType::Struct(vec![
736 ("pool_index".to_string(), DynType::U8),
737 ("quantity_is_input".to_string(), DynType::Bool),
738 ("quantity_is_collateral".to_string(), DynType::Bool),
739 ])),
740 ),
741 (
742 "SanctumS".to_string(),
743 Some(DynType::Struct(vec![
744 ("src_lst_value_calc_accs".to_string(), DynType::U8),
745 ("dst_lst_value_calc_accs".to_string(), DynType::U8),
746 ("src_lst_index".to_string(), DynType::U32),
747 ("dst_lst_index".to_string(), DynType::U32),
748 ])),
749 ),
750 (
751 "SanctumSAddLiquidity".to_string(),
752 Some(DynType::Struct(vec![
753 ("lst_value_calc_accs".to_string(), DynType::U8),
754 ("lst_index".to_string(), DynType::U32),
755 ])),
756 ),
757 (
758 "SanctumSRemoveLiquidity".to_string(),
759 Some(DynType::Struct(vec![
760 ("lst_value_calc_accs".to_string(), DynType::U8),
761 ("lst_index".to_string(), DynType::U32),
762 ])),
763 ),
764 ("RaydiumCP".to_string(), None),
765 (
766 "WhirlpoolSwapV2".to_string(),
767 Some(DynType::Struct(vec![
768 ("a_to_b".to_string(), DynType::Bool),
769 (
770 "remaining_accounts_info".to_string(),
771 DynType::Struct(vec![(
772 "slices".to_string(),
773 DynType::Array(Box::new(DynType::Struct(vec![(
774 "remaining_accounts_slice".to_string(),
775 DynType::Struct(vec![
776 ("accounts_type".to_string(), DynType::U8),
777 ("length".to_string(), DynType::U8),
778 ]),
779 )]))),
780 )]),
781 ),
782 ])),
783 ),
784 ("OneIntro".to_string(), None),
785 ("PumpdotfunWrappedBuy".to_string(), None),
786 ("PumpdotfunWrappedSell".to_string(), None),
787 ("PerpsV2".to_string(), None),
788 ("PerpsV2AddLiquidity".to_string(), None),
789 ("PerpsV2RemoveLiquidity".to_string(), None),
790 ("MoonshotWrappedBuy".to_string(), None),
791 ("MoonshotWrappedSell".to_string(), None),
792 ("StabbleStableSwap".to_string(), None),
793 ("StabbleWeightedSwap".to_string(), None),
794 (
795 "Obric".to_string(),
796 Some(DynType::Struct(vec![(
797 "x_to_y".to_string(),
798 DynType::Bool,
799 )])),
800 ),
801 ("FoxBuyFromEstimatedCost".to_string(), None),
802 (
803 "FoxClaimPartial".to_string(),
804 Some(DynType::Struct(vec![(
805 "is_y".to_string(),
806 DynType::Bool,
807 )])),
808 ),
809 (
810 "SolFi".to_string(),
811 Some(DynType::Struct(vec![(
812 "is_quote_to_base".to_string(),
813 DynType::Bool,
814 )])),
815 ),
816 ("SolayerDelegateNoInit".to_string(), None),
817 ("SolayerUndelegateNoInit".to_string(), None),
818 (
819 "TokenMill".to_string(),
820 Some(DynType::Struct(vec![(
821 "side".to_string(),
822 DynType::Enum(vec![
823 ("Bid".to_string(), None),
824 ("Ask".to_string(), None),
825 ]),
826 )])),
827 ),
828 ("DaosFunBuy".to_string(), None),
829 ("DaosFunSell".to_string(), None),
830 ("ZeroFi".to_string(), None),
831 ("StakeDexWithdrawWrappedSol".to_string(), None),
832 ("VirtualsBuy".to_string(), None),
833 ("VirtualsSell".to_string(), None),
834 (
835 "Peren".to_string(),
836 Some(DynType::Struct(vec![
837 ("in_index".to_string(), DynType::U8),
838 ("out_index".to_string(), DynType::U8),
839 ])),
840 ),
841 ("PumpdotfunAmmBuy".to_string(), None),
842 ("PumpdotfunAmmSell".to_string(), None),
843 ("Gamma".to_string(), None),
844 ]),
845 ),
846 ("Percent".to_string(), DynType::U8),
847 ("InputIndex".to_string(), DynType::U8),
848 ("OutputIndex".to_string(), DynType::U8),
849 ]))),
850 },
851 ParamInput {
852 name: "InAmount".to_string(),
853 param_type: DynType::U64,
854 },
855 ParamInput {
856 name: "QuotedOutAmount".to_string(),
857 param_type: DynType::U64,
858 },
859 ParamInput {
860 name: "SlippageBps".to_string(),
861 param_type: DynType::U16,
862 },
863 ParamInput {
864 name: "PlatformFeeBps".to_string(),
865 param_type: DynType::U8,
866 },
867 ],
868 accounts_names: vec![
869 "TokenProgram".to_string(),
870 "UserTransferAuthority".to_string(),
871 "UserSourceTokenAccount".to_string(),
872 "UserDestinationTokenAccount".to_string(),
873 "DestinationTokenAccount".to_string(),
874 "PlatformFeeAccount".to_string(),
875 "EventAuthority".to_string(),
876 "Program".to_string(),
877 "test8".to_string(),
878 "test9".to_string(),
879 ],
880 };
881
882 let result = decode_instructions_batch(ix_signature, &instructions, true)
883 .context("decode failed")
884 .unwrap();
885
886 let mut file = File::create("decoded_instructions.parquet").unwrap();
888 let mut writer =
889 parquet::arrow::ArrowWriter::try_new(&mut file, result.schema(), None).unwrap();
890 writer.write(&result).unwrap();
891 writer.close().unwrap();
892 }
893
894 #[test]
895 #[ignore]
896 fn test_decode_logs_with_real_data() {
897 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
898
899 let builder =
900 ParquetRecordBatchReaderBuilder::try_new(File::open("logs.parquet").unwrap()).unwrap();
901 let mut reader = builder.build().unwrap();
902 let logs = reader.next().unwrap().unwrap();
903
904 let signature = LogSignature {
905 params: vec![
906 ParamInput {
907 name: "whirlpool".to_string(),
908 param_type: DynType::FixedArray(Box::new(DynType::U8), 32),
909 },
910 ParamInput {
911 name: "a_to_b".to_string(),
912 param_type: DynType::Bool,
913 },
914 ParamInput {
915 name: "pre_sqrt_price".to_string(),
916 param_type: DynType::U128,
917 },
918 ParamInput {
919 name: "post_sqrt_price".to_string(),
920 param_type: DynType::U128,
921 },
922 ParamInput {
923 name: "x".to_string(),
924 param_type: DynType::U64,
925 },
926 ParamInput {
927 name: "input_amount".to_string(),
928 param_type: DynType::U64,
929 },
930 ParamInput {
931 name: "output_amount".to_string(),
932 param_type: DynType::U64,
933 },
934 ParamInput {
935 name: "input_transfer_fee".to_string(),
936 param_type: DynType::U64,
937 },
938 ParamInput {
939 name: "output_transfer_fee".to_string(),
940 param_type: DynType::U64,
941 },
942 ParamInput {
943 name: "lp_fee".to_string(),
944 param_type: DynType::U64,
945 },
946 ParamInput {
947 name: "protocol_fee".to_string(),
948 param_type: DynType::U64,
949 },
950 ],
951 };
952
953 let result = decode_logs_batch(signature, &logs, true)
954 .context("decode failed")
955 .unwrap();
956
957 let mut file = File::create("decoded_logs.parquet").unwrap();
959 let mut writer =
960 parquet::arrow::ArrowWriter::try_new(&mut file, result.schema(), None).unwrap();
961 writer.write(&result).unwrap();
962 writer.close().unwrap();
963 }
964
965 #[test]
966 #[ignore]
967 fn test_instruction_signature_to_arrow_schema() {
968 let signature = InstructionSignature {
970 discriminator: vec![],
971 params: vec![
972 ParamInput {
973 name: "amount".to_string(),
974 param_type: DynType::U64,
975 },
976 ParamInput {
977 name: "is_valid".to_string(),
978 param_type: DynType::Bool,
979 },
980 ParamInput {
981 name: "amm".to_string(),
982 param_type: DynType::FixedArray(Box::new(DynType::U8), 32),
983 },
984 ],
985 accounts_names: vec!["source".to_string(), "destination".to_string()],
986 };
987
988 let schema = instruction_signature_to_arrow_schema(&signature).unwrap();
990
991 assert_eq!(schema.fields().len(), 5); let amount_field = schema.field_with_name("amount").unwrap();
996 assert_eq!(amount_field.name(), "amount");
997 assert!(amount_field.is_nullable());
998
999 let is_valid_field = schema.field_with_name("is_valid").unwrap();
1000 assert_eq!(is_valid_field.name(), "is_valid");
1001 assert!(is_valid_field.is_nullable());
1002
1003 let amm_field = schema.field_with_name("amm").unwrap();
1004 assert_eq!(amm_field.name(), "amm");
1005 assert!(amm_field.is_nullable());
1006
1007 let source_field = schema.field_with_name("source").unwrap();
1009 assert_eq!(source_field.name(), "source");
1010 assert_eq!(source_field.data_type(), &DataType::Binary);
1011 assert!(source_field.is_nullable());
1012
1013 let dest_field = schema.field_with_name("destination").unwrap();
1014 assert_eq!(dest_field.name(), "destination");
1015 assert_eq!(dest_field.data_type(), &DataType::Binary);
1016 assert!(dest_field.is_nullable());
1017 }
1018}